# Source for file Balancer.php

Documentation is available at Balancer.php

1. <?php
2. /**
3. * Copyright (c) 2007-2009, Conduit Internet Technologies, Inc.
4. * All rights reserved.
5. *
6. * Redistribution and use in source and binary forms, with or without
7. * modification, are permitted provided that the following conditions are met:
8. *
9. * - Redistributions of source code must retain the above copyright notice,
10. * this list of conditions and the following disclaimer.
11. * - Redistributions in binary form must reproduce the above copyright
12. * notice, this list of conditions and the following disclaimer in the
13. * documentation and/or other materials provided with the distribution.
14. * - Neither the name of Conduit Internet Technologies, Inc. nor the names of
15. * its contributors may be used to endorse or promote products derived from
16. * this software without specific prior written permission.
17. *
18. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
19. * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20. * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21. * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
22. * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
23. * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24. * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
25. * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
26. * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
27. * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
28. * POSSIBILITY OF SUCH DAMAGE.
29. *
30. * @copyright Copyright 2007-2009 Conduit Internet Technologies, Inc. (http://conduit-it.com)
31. * @license New BSD (http://solr-php-client.googlecode.com/svn/trunk/COPYING)
32. * @version $Id: Balancer.php 15 2009-08-04 17:53:08Z donovan.jimenez $
33. *
34. * @package Apache
35. * @subpackage Solr
36. * @author Donovan Jimenez <djimenez@conduit-it.com>, Dan Wolfe
37. */
39. // See Issue #1 (http://code.google.com/p/solr-php-client/issues/detail?id=1)
40. // Doesn't follow typical include path conventions, but is more convenient for users
41. require_once(dirname(dirname(__FILE__)) . '/Service.php');
43. /**
44. * Reference Implementation for using multiple Solr services in a distribution. Functionality
45. * includes:
46. * routing of read / write operations
47. * failover (on selection) for multiple read servers
48. */
49. class Apache_Solr_Service_Balancer
50. {
51. /**
52. * SVN Revision meta data for this class
53. */
54. const SVN_REVISION = '$Revision: 15 $';
56. /**
57. * SVN ID meta data for this class
58. */
59. const SVN_ID = '$Id: Balancer.php 15 2009-08-04 17:53:08Z donovan.jimenez $';
61. protected $_createDocuments = true;
63. protected $_readableServices = array();
64. protected $_writeableServices = array();
66. protected $_currentReadService = null;
67. protected $_currentWriteService = null;
69. protected $_readPingTimeout = 2;
70. protected $_writePingTimeout = 4;
72. // Configuration for server selection backoff intervals
73. protected $_useBackoff = false; // Set to true to use more resillient write server selection
74. protected $_backoffLimit = 600; // 10 minute default maximum
75. protected $_backoffEscalation = 2.0; // Rate at which to increase backoff period
76. protected $_defaultBackoff = 2.0; // Default backoff interval
78. /**
79. * Escape a value for special query characters such as ':', '(', ')', '*', '?', etc.
80. *
81. * NOTE: inside a phrase fewer characters need escaped, use {@link Apache_Solr_Service::escapePhrase()} instead
82. *
83. * @param string $value
84. * @return string
85. */
86. static public function escape($value)
87. {
88. return Apache_Solr_Service::escape($value);
89. }
91. /**
92. * Escape a value meant to be contained in a phrase for special query characters
93. *
94. * @param string $value
95. * @return string
96. */
97. static public function escapePhrase($value)
98. {
99. return Apache_Solr_Service::escapePhrase($value);
100. }
102. /**
103. * Convenience function for creating phrase syntax from a value
104. *
105. * @param string $value
106. * @return string
107. */
108. static public function phrase($value)
109. {
110. return Apache_Solr_Service::phrase($value);
111. }
113. /**
114. * Constructor. Takes arrays of read and write service instances or descriptions
115. *
116. * @param array $readableServices
117. * @param array $writeableServices
118. */
119. public function __construct($readableServices = array(), $writeableServices = array())
120. {
121. //setup readable services
122. foreach ($readableServices as $service)
123. {
124. $this->addReadService($service);
125. }
127. //setup writeable services
128. foreach ($writeableServices as $service)
129. {
130. $this->addWriteService($service);
131. }
132. }
134. public function setReadPingTimeout($timeout)
135. {
136. $this->_readPingTimeout = $timeout;
137. }
139. public function setWritePingTimeout($timeout)
140. {
141. $this->_writePingTimeout = $timeout;
142. }
144. public function setUseBackoff($enable)
145. {
146. $this->_useBackoff = $enable;
147. }
149. /**
150. * Generates a service ID
151. *
152. * @param string $host
153. * @param integer $port
154. * @param string $path
155. * @return string
156. */
157. protected function _getServiceId($host, $port, $path)
158. {
159. return $host . ':' . $port . $path;
160. }
162. /**
163. * Adds a service instance or service descriptor (if it is already
164. * not added)
165. *
166. * @param mixed $service
167. *
168. * @throws Exception If service descriptor is not valid
169. */
170. public function addReadService($service)
171. {
172. if ($service instanceof Apache_Solr_Service)
173. {
174. $id = $this->_getServiceId($service->getHost(), $service->getPort(), $service->getPath());
176. $this->_readableServices[$id] = $service;
177. }
178. else if (is_array($service))
179. {
180. if (isset($service['host']) && isset($service['port']) && isset($service['path']))
181. {
182. $id = $this->_getServiceId((string)$service['host'], (int)$service['port'], (string)$service['path']);
184. $this->_readableServices[$id] = $service;
185. }
186. else
187. {
188. throw new Exception('A Readable Service description array does not have all required elements of host, port, and path');
189. }
190. }
191. }
193. /**
194. * Removes a service instance or descriptor from the available services
195. *
196. * @param mixed $service
197. *
198. * @throws Exception If service descriptor is not valid
199. */
200. public function removeReadService($service)
201. {
202. $id = '';
204. if ($service instanceof Apache_Solr_Service)
205. {
206. $id = $this->_getServiceId($service->getHost(), $service->getPort(), $service->getPath());
207. }
208. else if (is_array($service))
209. {
210. if (isset($service['host']) && isset($service['port']) && isset($service['path']))
211. {
212. $id = $this->_getServiceId((string)$service['host'], (int)$service['port'], (string)$service['path']);
213. }
214. else
215. {
216. throw new Exception('A Readable Service description array does not have all required elements of host, port, and path');
217. }
218. }
219. else if (is_string($service))
220. {
221. $id = $service;
222. }
224. if ($id && isset($this->_readableServices[$id]))
225. {
226. unset($this->_readableServices[$id]);
227. }
228. }
230. /**
231. * Adds a service instance or service descriptor (if it is already
232. * not added)
233. *
234. * @param mixed $service
235. *
236. * @throws Exception If service descriptor is not valid
237. */
238. public function addWriteService($service)
239. {
240. if ($service instanceof Apache_Solr_Service)
241. {
242. $id = $this->_getServiceId($service->getHost(), $service->getPort(), $service->getPath());
244. $this->_writeableServices[$id] = $service;
245. }
246. else if (is_array($service))
247. {
248. if (isset($service['host']) && isset($service['port']) && isset($service['path']))
249. {
250. $id = $this->_getServiceId((string)$service['host'], (int)$service['port'], (string)$service['path']);
252. $this->_writeableServices[$id] = $service;
253. }
254. else
255. {
256. throw new Exception('A Writeable Service description array does not have all required elements of host, port, and path');
257. }
258. }
259. }
261. /**
262. * Removes a service instance or descriptor from the available services
263. *
264. * @param mixed $service
265. *
266. * @throws Exception If service descriptor is not valid
267. */
268. public function removeWriteService($service)
269. {
270. $id = '';
272. if ($service instanceof Apache_Solr_Service)
273. {
274. $id = $this->_getServiceId($service->getHost(), $service->getPort(), $service->getPath());
275. }
276. else if (is_array($service))
277. {
278. if (isset($service['host']) && isset($service['port']) && isset($service['path']))
279. {
280. $id = $this->_getServiceId((string)$service['host'], (int)$service['port'], (string)$service['path']);
281. }
282. else
283. {
284. throw new Exception('A Readable Service description array does not have all required elements of host, port, and path');
285. }
286. }
287. else if (is_string($service))
288. {
289. $id = $service;
290. }
292. if ($id && isset($this->_writeableServices[$id]))
293. {
294. unset($this->_writeableServices[$id]);
295. }
296. }
298. /**
299. * Iterate through available read services and select the first with a ping
300. * that satisfies configured timeout restrictions (or the default)
301. *
302. * @return Apache_Solr_Service
303. *
304. * @throws Exception If there are no read services that meet requirements
305. */
306. protected function _selectReadService($forceSelect = false)
307. {
308. if (!$this->_currentReadService || !isset($this->_readableServices[$this->_currentReadService]) || $forceSelect)
309. {
310. if ($this->_currentReadService && isset($this->_readableServices[$this->_currentReadService]) && $forceSelect)
311. {
312. // we probably had a communication error, ping the current read service, remove it if it times out
313. if ($this->_readableServices[$this->_currentReadService]->ping($this->_readPingTimeout) === false)
314. {
315. $this->removeReadService($this->_currentReadService);
316. }
317. }
319. if (count($this->_readableServices))
320. {
321. // select one of the read services at random
322. $ids = array_keys($this->_readableServices);
324. $id = $ids[rand(0, count($ids) - 1)];
325. $service = $this->_readableServices[$id];
327. if (is_array($service))
328. {
329. //convert the array definition to a client object
330. $service = new Apache_Solr_Service($service['host'], $service['port'], $service['path']);
331. $this->_readableServices[$id] = $service;
332. }
334. $service->setCreateDocuments($this->_createDocuments);
335. $this->_currentReadService = $id;
336. }
337. else
338. {
339. throw new Exception('No read services were available');
340. }
341. }
343. return $this->_readableServices[$this->_currentReadService];
344. }
346. /**
347. * Iterate through available write services and select the first with a ping
348. * that satisfies configured timeout restrictions (or the default)
349. *
350. * @return Apache_Solr_Service
351. *
352. * @throws Exception If there are no write services that meet requirements
353. */
354. protected function _selectWriteService($forceSelect = false)
355. {
356. if($this->_useBackoff)
357. {
358. return $this->_selectWriteServiceSafe($forceSelect);
359. }
361. if (!$this->_currentWriteService || !isset($this->_writeableServices[$this->_currentWriteService]) || $forceSelect)
362. {
363. if ($this->_currentWriteService && isset($this->_writeableServices[$this->_currentWriteService]) && $forceSelect)
364. {
365. // we probably had a communication error, ping the current read service, remove it if it times out
366. if ($this->_writeableServices[$this->_currentWriteService]->ping($this->_writePingTimeout) === false)
367. {
368. $this->removeWriteService($this->_currentWriteService);
369. }
370. }
372. if (count($this->_writeableServices))
373. {
374. // select one of the read services at random
375. $ids = array_keys($this->_writeableServices);
377. $id = $ids[rand(0, count($ids) - 1)];
378. $service = $this->_writeableServices[$id];
380. if (is_array($service))
381. {
382. //convert the array definition to a client object
383. $service = new Apache_Solr_Service($service['host'], $service['port'], $service['path']);
384. $this->_writeableServices[$id] = $service;
385. }
387. $this->_currentWriteService = $id;
388. }
389. else
390. {
391. throw new Exception('No write services were available');
392. }
393. }
395. return $this->_writeableServices[$this->_currentWriteService];
396. }
398. /**
399. * Iterate through available write services and select the first with a ping
400. * that satisfies configured timeout restrictions (or the default). The
401. * timeout period will increase until a connection is made or the limit is
402. * reached. This will allow for increased reliability with heavily loaded
403. * server(s).
404. *
405. * @return Apache_Solr_Service
406. *
407. * @throws Exception If there are no write services that meet requirements
408. */
410. protected function _selectWriteServiceSafe($forceSelect = false)
411. {
412. if (!$this->_currentWriteService || !isset($this->_writeableServices[$this->_currentWriteService]) || $forceSelect)
413. {
414. if (count($this->_writeableServices))
415. {
416. $backoff = $this->_defaultBackoff;
418. do {
419. // select one of the read services at random
420. $ids = array_keys($this->_writeableServices);
422. $id = $ids[rand(0, count($ids) - 1)];
423. $service = $this->_writeableServices[$id];
425. if (is_array($service))
426. {
427. //convert the array definition to a client object
428. $service = new Apache_Solr_Service($service['host'], $service['port'], $service['path']);
429. $this->_writeableServices[$id] = $service;
430. }
432. $this->_currentWriteService = $id;
434. $backoff *= $this->_backoffEscalation;
436. if($backoff > $this->_backoffLimit)
437. {
438. throw new Exception('No write services were available. All timeouts exceeded.');
439. }
441. } while($this->_writeableServices[$this->_currentWriteService]->ping($backoff) === false);
442. }
443. else
444. {
445. throw new Exception('No write services were available');
446. }
447. }
449. return $this->_writeableServices[$this->_currentWriteService];
450. }
452. public function setCreateDocuments($createDocuments)
453. {
454. $this->_createDocuments = (bool) $createDocuments;
456. if ($this->_currentReadService)
457. {
458. $service = $this->_selectReadService();
459. $service->setCreateDocuments($createDocuments);
460. }
461. }
463. public function getCreateDocuments()
464. {
465. return $this->_createDocuments;
466. }
468. /**
469. * Raw Add Method. Takes a raw post body and sends it to the update service. Post body
470. * should be a complete and well formed "add" xml document.
471. *
472. * @param string $rawPost
473. * @return Apache_Solr_Response
474. *
475. * @throws Exception If an error occurs during the service call
476. */
477. public function add($rawPost)
478. {
479. $service = $this->_selectWriteService();
481. do
482. {
483. try
484. {
485. return $service->add($rawPost);
486. }
487. catch (Exception $e)
488. {
489. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
490. {
491. throw $e;
492. }
493. }
495. $service = $this->_selectWriteService(true);
496. } while ($service);
498. return false;
499. }
501. /**
502. * Add a Solr Document to the index
503. *
504. * @param Apache_Solr_Document $document
505. * @param boolean $allowDups
506. * @param boolean $overwritePending
507. * @param boolean $overwriteCommitted
508. * @return Apache_Solr_Response
509. *
510. * @throws Exception If an error occurs during the service call
511. */
512. public function addDocument(Apache_Solr_Document $document, $allowDups = false, $overwritePending = true, $overwriteCommitted = true)
513. {
514. $service = $this->_selectWriteService();
516. do
517. {
518. try
519. {
520. return $service->addDocument($document, $allowDups, $overwritePending, $overwriteCommitted);
521. }
522. catch (Exception $e)
523. {
524. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
525. {
526. throw $e;
527. }
528. }
530. $service = $this->_selectWriteService(true);
531. } while ($service);
533. return false;
534. }
536. /**
537. * Add an array of Solr Documents to the index all at once
538. *
539. * @param array $documents Should be an array of Apache_Solr_Document instances
540. * @param boolean $allowDups
541. * @param boolean $overwritePending
542. * @param boolean $overwriteCommitted
543. * @return Apache_Solr_Response
544. *
545. * @throws Exception If an error occurs during the service call
546. */
547. public function addDocuments($documents, $allowDups = false, $overwritePending = true, $overwriteCommitted = true)
548. {
549. $service = $this->_selectWriteService();
551. do
552. {
553. try
554. {
555. return $service->addDocuments($documents, $allowDups, $overwritePending, $overwriteCommitted);
556. }
557. catch (Exception $e)
558. {
559. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
560. {
561. throw $e;
562. }
563. }
565. $service = $this->_selectWriteService(true);
566. } while ($service);
568. return false;
569. }
571. /**
572. * Send a commit command. Will be synchronous unless both wait parameters are set
573. * to false.
574. *
575. * @param boolean $waitFlush
576. * @param boolean $waitSearcher
577. * @return Apache_Solr_Response
578. *
579. * @throws Exception If an error occurs during the service call
580. */
581. public function commit($optimize = true, $waitFlush = true, $waitSearcher = true, $timeout = 3600)
582. {
583. $service = $this->_selectWriteService();
585. do
586. {
587. try
588. {
589. return $service->commit($optimize, $waitFlush, $waitSearcher, $timeout);
590. }
591. catch (Exception $e)
592. {
593. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
594. {
595. throw $e;
596. }
597. }
599. $service = $this->_selectWriteService(true);
600. } while ($service);
602. return false;
603. }
605. /**
606. * Raw Delete Method. Takes a raw post body and sends it to the update service. Body should be
607. * a complete and well formed "delete" xml document
608. *
609. * @param string $rawPost
610. * @return Apache_Solr_Response
611. *
612. * @throws Exception If an error occurs during the service call
613. */
614. public function delete($rawPost)
615. {
616. $service = $this->_selectWriteService();
618. do
619. {
620. try
621. {
622. return $service->delete($rawPost);
623. }
624. catch (Exception $e)
625. {
626. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
627. {
628. throw $e;
629. }
630. }
632. $service = $this->_selectWriteService(true);
633. } while ($service);
635. return false;
636. }
638. /**
639. * Create a delete document based on document ID
640. *
641. * @param string $id
642. * @param boolean $fromPending
643. * @param boolean $fromCommitted
644. * @return Apache_Solr_Response
645. *
646. * @throws Exception If an error occurs during the service call
647. */
648. public function deleteById($id, $fromPending = true, $fromCommitted = true)
649. {
650. $service = $this->_selectWriteService();
652. do
653. {
654. try
655. {
656. return $service->deleteById($id, $fromPending, $fromCommitted);
657. }
658. catch (Exception $e)
659. {
660. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
661. {
662. throw $e;
663. }
664. }
666. $service = $this->_selectWriteService(true);
667. } while ($service);
669. return false;
670. }
672. /**
673. * Create a delete document based on a query and submit it
674. *
675. * @param string $rawQuery
676. * @param boolean $fromPending
677. * @param boolean $fromCommitted
678. * @return Apache_Solr_Response
679. *
680. * @throws Exception If an error occurs during the service call
681. */
682. public function deleteByQuery($rawQuery, $fromPending = true, $fromCommitted = true)
683. {
684. $service = $this->_selectWriteService();
686. do
687. {
688. try
689. {
690. return $service->deleteByQuery($rawQuery, $fromPending, $fromCommitted);
691. }
692. catch (Exception $e)
693. {
694. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
695. {
696. throw $e;
697. }
698. }
700. $service = $this->_selectWriteService(true);
701. } while ($service);
703. return false;
704. }
706. /**
707. * Send an optimize command. Will be synchronous unless both wait parameters are set
708. * to false.
709. *
710. * @param boolean $waitFlush
711. * @param boolean $waitSearcher
712. * @return Apache_Solr_Response
713. *
714. * @throws Exception If an error occurs during the service call
715. */
716. public function optimize($waitFlush = true, $waitSearcher = true)
717. {
718. $service = $this->_selectWriteService();
720. do
721. {
722. try
723. {
724. return $service->optimize($waitFlush, $waitSearcher);
725. }
726. catch (Exception $e)
727. {
728. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
729. {
730. throw $e;
731. }
732. }
734. $service = $this->_selectWriteService(true);
735. } while ($service);
737. return false;
738. }
740. /**
741. * Simple Search interface
742. *
743. * @param string $query The raw query string
744. * @param int $offset The starting offset for result documents
745. * @param int $limit The maximum number of result documents to return
746. * @param array $params key / value pairs for query parameters, use arrays for multivalued parameters
747. * @return Apache_Solr_Response
748. *
749. * @throws Exception If an error occurs during the service call
750. */
751. public function search($query, $offset = 0, $limit = 10, $params = array())
752. {
753. $service = $this->_selectReadService();
755. do
756. {
757. try
758. {
759. return $service->search($query, $offset, $limit, $params);
760. }
761. catch (Exception $e)
762. {
763. if ($e->getCode() != 0) //IF NOT COMMUNICATION ERROR
764. {
765. throw $e;
766. }
767. }
769. $service = $this->_selectReadService(true);
770. } while ($service);
772. return false;
773. }
774. }

Documentation generated on Mon, 09 Nov 2009 18:15:41 -0500 by phpDocumentor 1.4.2