sync_persist.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694
  1. /** BEGIN COPYRIGHT BLOCK
  2. ent sync_srch_refresh_pre_op(Slapi_PBlock *pb);
  3. * This Program is free software; you can redistribute it and/or modify it under
  4. * the terms of the GNU General Public License as published by the Free Software
  5. * Foundation; version 2 of the License.
  6. *
  7. * This Program is distributed in the hope that it will be useful, but WITHOUT
  8. * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  9. * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
  10. *
  11. * You should have received a copy of the GNU General Public License along with
  12. * this Program; if not, write to the Free Software Foundation, Inc., 59 Temple
  13. * Place, Suite 330, Boston, MA 02111-1307 USA.
  14. *
  15. * In addition, as a special exception, Red Hat, Inc. gives You the additional
  16. * right to link the code of this Program with code not covered under the GNU
  17. * General Public License ("Non-GPL Code") and to distribute linked combinations
  18. * including the two, subject to the limitations in this paragraph. Non-GPL Code
  19. * permitted under this exception must only link to the code of this Program
  20. * through those well defined interfaces identified in the file named EXCEPTION
  21. * found in the source code files (the "Approved Interfaces"). The files of
  22. * Non-GPL Code may instantiate templates or use macros or inline functions from
  23. * the Approved Interfaces without causing the resulting work to be covered by
  24. * the GNU General Public License. Only Red Hat, Inc. may make changes or
  25. * additions to the list of Approved Interfaces. You must obey the GNU General
  26. * Public License in all respects for all of the Program code and other code used
  27. * in conjunction with the Program except the Non-GPL Code covered by this
  28. * exception. If you modify this file, you may extend this exception to your
  29. * version of the file, but you are not obligated to do so. If you do not wish to
  30. * provide this exception without modification, you must delete this exception
  31. * statement from your version and license this file solely under the GPL without
  32. * exception.
  33. *
  34. *
  35. * Copyright (C) 2013 Red Hat, Inc.
  36. * All rights reserved.
  37. * END COPYRIGHT BLOCK **/
  38. #include "sync.h"
  39. /* Main list of established persistent synchronizaton searches */
  40. static SyncRequestList *sync_request_list = NULL;
  41. /*
  42. * Convenience macros for locking the list of persistent searches
  43. */
  44. #define SYNC_LOCK_READ() slapi_rwlock_rdlock(sync_request_list->sync_req_rwlock)
  45. #define SYNC_UNLOCK_READ() slapi_rwlock_unlock(sync_request_list->sync_req_rwlock)
  46. #define SYNC_LOCK_WRITE() slapi_rwlock_wrlock(sync_request_list->sync_req_rwlock)
  47. #define SYNC_UNLOCK_WRITE() slapi_rwlock_unlock(sync_request_list->sync_req_rwlock)
  48. /*
  49. * Convenience macro for checking if the Content Synchronization subsystem has
  50. * been initialized.
  51. */
  52. #define SYNC_IS_INITIALIZED() (sync_request_list != NULL)
  53. static int sync_add_request( SyncRequest *req );
  54. static void sync_remove_request( SyncRequest *req );
  55. static SyncRequest *sync_request_alloc();
  56. void sync_queue_change( Slapi_Entry *e, Slapi_Entry *eprev, ber_int_t chgtype );
  57. static void sync_send_results( void *arg );
  58. static void sync_request_wakeup_all();
  59. static void sync_node_free( SyncQueueNode **node );
  60. static int sync_acquire_connection (Slapi_Connection *conn);
  61. static int sync_release_connection (Slapi_PBlock *pb, Slapi_Connection *conn, Slapi_Operation *op, int release);
  62. int sync_add_persist_post_op(Slapi_PBlock *pb)
  63. {
  64. Slapi_Entry *e;
  65. if ( !SYNC_IS_INITIALIZED()) {
  66. return(0);
  67. }
  68. slapi_pblock_get(pb, SLAPI_ENTRY_POST_OP, &e);
  69. sync_queue_change(e, NULL, LDAP_REQ_ADD);
  70. return( 0 );
  71. }
  72. int sync_del_persist_post_op(Slapi_PBlock *pb)
  73. {
  74. Slapi_Entry *e;
  75. if ( !SYNC_IS_INITIALIZED()) {
  76. return(0);
  77. }
  78. slapi_pblock_get(pb, SLAPI_ENTRY_PRE_OP, &e);
  79. sync_queue_change(e, NULL, LDAP_REQ_DELETE);
  80. return( 0 );
  81. }
  82. int sync_mod_persist_post_op(Slapi_PBlock *pb)
  83. {
  84. Slapi_Entry *e;
  85. if ( !SYNC_IS_INITIALIZED()) {
  86. return(0);
  87. }
  88. slapi_pblock_get(pb, SLAPI_ENTRY_POST_OP, &e);
  89. sync_queue_change(e, NULL, LDAP_REQ_MODIFY);
  90. return( 0 );
  91. }
  92. int sync_modrdn_persist_post_op(Slapi_PBlock *pb)
  93. {
  94. Slapi_Entry *e, *e_prev;
  95. if ( !SYNC_IS_INITIALIZED()) {
  96. return(0);
  97. }
  98. slapi_pblock_get(pb, SLAPI_ENTRY_POST_OP, &e);
  99. slapi_pblock_get(pb, SLAPI_ENTRY_PRE_OP, &e_prev);
  100. sync_queue_change(e, e_prev, LDAP_REQ_MODRDN);
  101. return( 0 );
  102. }
  103. void
  104. sync_queue_change( Slapi_Entry *e, Slapi_Entry *eprev, ber_int_t chgtype )
  105. {
  106. SyncRequest *req = NULL;
  107. SyncQueueNode *node = NULL;
  108. int matched = 0;
  109. int prev_match = 0;
  110. int cur_match = 0;
  111. if ( !SYNC_IS_INITIALIZED()) {
  112. return;
  113. }
  114. if ( NULL == e ) {
  115. /* For now, some backends such as the chaining backend do not provide a post-op entry */
  116. return;
  117. }
  118. SYNC_LOCK_READ();
  119. for ( req = sync_request_list->sync_req_head; NULL != req; req = req->req_next ) {
  120. Slapi_DN *base = NULL;
  121. int scope;
  122. Slapi_Operation *op;
  123. /* Skip the nodes that have no more active operation
  124. */
  125. slapi_pblock_get( req->req_pblock, SLAPI_OPERATION, &op );
  126. if ( op == NULL || slapi_op_abandoned( req->req_pblock ) ) {
  127. continue;
  128. }
  129. slapi_pblock_get( req->req_pblock, SLAPI_SEARCH_TARGET_SDN, &base );
  130. slapi_pblock_get( req->req_pblock, SLAPI_SEARCH_SCOPE, &scope );
  131. if (NULL == base) {
  132. base = slapi_sdn_new_dn_byref(req->req_orig_base);
  133. slapi_pblock_set(req->req_pblock, SLAPI_SEARCH_TARGET_SDN, base);
  134. }
  135. /*
  136. * See if the entry meets the scope and filter criteria.
  137. * We cannot do the acl check here as this thread
  138. * would then potentially clash with the ps_send_results()
  139. * thread on the aclpb in ps->req_pblock.
  140. * By avoiding the acl check in this thread, and leaving all the acl
  141. * checking to the ps_send_results() thread we avoid
  142. * the req_pblock contention problem.
  143. * The lesson here is "Do not give multiple threads arbitary access
  144. * to the same pblock" this kind of muti-threaded access
  145. * to the same pblock must be done carefully--there is currently no
  146. * generic satisfactory way to do this.
  147. */
  148. /* if the change is a modrdn then we need to check if the entry was
  149. * moved into scope, out of scope, or stays in scope
  150. */
  151. if (chgtype == LDAP_REQ_MODRDN)
  152. prev_match = slapi_sdn_scope_test( slapi_entry_get_sdn_const(eprev), base, scope ) &&
  153. ( 0 == slapi_vattr_filter_test( req->req_pblock, eprev, req->req_filter, 0 /* verify_access */ ));
  154. cur_match = slapi_sdn_scope_test( slapi_entry_get_sdn_const(e), base, scope ) &&
  155. ( 0 == slapi_vattr_filter_test( req->req_pblock, e, req->req_filter, 0 /* verify_access */ ));
  156. if (prev_match || cur_match) {
  157. SyncQueueNode *pOldtail;
  158. /* The scope and the filter match - enqueue it */
  159. matched++;
  160. node = (SyncQueueNode *)slapi_ch_calloc( 1, sizeof( SyncQueueNode ));
  161. node->sync_entry = slapi_entry_dup( e );
  162. if ( chgtype == LDAP_REQ_MODRDN) {
  163. if (prev_match && cur_match)
  164. node->sync_chgtype = LDAP_REQ_MODIFY;
  165. else if (prev_match)
  166. node->sync_chgtype = LDAP_REQ_DELETE;
  167. else
  168. node->sync_chgtype = LDAP_REQ_ADD;
  169. } else {
  170. node->sync_chgtype = chgtype;
  171. }
  172. /* Put it on the end of the list for this sync search */
  173. PR_Lock( req->req_lock );
  174. pOldtail = req->ps_eq_tail;
  175. req->ps_eq_tail = node;
  176. if ( NULL == req->ps_eq_head ) {
  177. req->ps_eq_head = req->ps_eq_tail;
  178. }
  179. else {
  180. pOldtail->sync_next = req->ps_eq_tail;
  181. }
  182. PR_Unlock( req->req_lock );
  183. }
  184. }
  185. SYNC_UNLOCK_READ();
  186. /* Were there any matches? */
  187. if ( matched ) {
  188. /* Notify update threads */
  189. sync_request_wakeup_all();
  190. slapi_log_error (SLAPI_LOG_TRACE, SYNC_PLUGIN_SUBSYSTEM, "sync search: enqueued entry "
  191. "\"%s\" on %d request listeners\n", slapi_entry_get_dn_const(e), matched );
  192. } else {
  193. slapi_log_error (SLAPI_LOG_TRACE, SYNC_PLUGIN_SUBSYSTEM, "sync search: entry "
  194. "\"%s\" not enqueued on any request search listeners\n", slapi_entry_get_dn_const(e) );
  195. }
  196. }
  197. /*
  198. * Initialize the list structure which contains the list
  199. * of established content sync persistent requests
  200. */
  201. int
  202. sync_persist_initialize (int argc, char **argv)
  203. {
  204. if ( !SYNC_IS_INITIALIZED()) {
  205. sync_request_list = (SyncRequestList *) slapi_ch_calloc( 1, sizeof( SyncRequestList ));
  206. if (( sync_request_list->sync_req_rwlock = slapi_new_rwlock()) == NULL ) {
  207. slapi_log_error (SLAPI_LOG_FATAL, SYNC_PLUGIN_SUBSYSTEM, "sync_persist_initialize: cannot initialize lock structure(1). ");
  208. return( -1 );
  209. }
  210. if (( sync_request_list->sync_req_cvarlock = PR_NewLock()) == NULL ) {
  211. slapi_log_error (SLAPI_LOG_FATAL, SYNC_PLUGIN_SUBSYSTEM, "sync_persist_initialize: cannot initialize lock structure(2). ");
  212. return( -1 );
  213. }
  214. if (( sync_request_list->sync_req_cvar = PR_NewCondVar( sync_request_list->sync_req_cvarlock )) == NULL ) {
  215. slapi_log_error (SLAPI_LOG_FATAL, SYNC_PLUGIN_SUBSYSTEM, "sync_persist_initialize: cannot initialize condition variable. ");
  216. return( -1 );
  217. }
  218. sync_request_list->sync_req_head = NULL;
  219. sync_request_list->sync_req_cur_persist = 0;
  220. sync_request_list->sync_req_max_persist = SYNC_MAX_CONCURRENT;
  221. if (argc > 0) {
  222. /* for now the only plugin arg is the max concurrent
  223. * persistent sync searches
  224. */
  225. sync_request_list->sync_req_max_persist = sync_number2int(argv[0]);
  226. if (sync_request_list->sync_req_max_persist == -1) {
  227. sync_request_list->sync_req_max_persist = SYNC_MAX_CONCURRENT;
  228. }
  229. }
  230. }
  231. return (0);
  232. }
  233. /*
  234. * Add the given pblock to the list of established sync searches.
  235. * Then, start a thread to send the results to the client as they
  236. * are dispatched by add, modify, and modrdn operations.
  237. */
  238. PRThread *
  239. sync_persist_add (Slapi_PBlock *pb)
  240. {
  241. SyncRequest *req = NULL;
  242. char *base;
  243. Slapi_Filter *filter;
  244. if ( SYNC_IS_INITIALIZED() && NULL != pb ) {
  245. /* Create the new node */
  246. req = sync_request_alloc();
  247. req->req_pblock = sync_pblock_copy(pb);
  248. slapi_pblock_get(pb, SLAPI_ORIGINAL_TARGET_DN, &base);
  249. req->req_orig_base = slapi_ch_strdup(base);
  250. slapi_pblock_get( pb, SLAPI_SEARCH_FILTER, &filter );
  251. req->req_filter = slapi_filter_dup(filter);
  252. /* Add it to the head of the list of persistent searches */
  253. if ( 0 == sync_add_request( req )) {
  254. /* Start a thread to send the results */
  255. req->req_tid = PR_CreateThread( PR_USER_THREAD, sync_send_results,
  256. (void *) req, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
  257. PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE );
  258. /* Checking if the thread is succesfully created and
  259. * if the thread is not created succesfully.... we send
  260. * error messages to the Log file
  261. */
  262. if(NULL == (req->req_tid)){
  263. int prerr;
  264. prerr = PR_GetError();
  265. slapi_log_error(SLAPI_LOG_FATAL, "Content Synchronization Search",
  266. "sync_persist_add function: failed to create persitent thread, error %d (%s)\n",
  267. prerr, slapi_pr_strerror(prerr));
  268. /* Now remove the ps from the list so call the function ps_remove */
  269. sync_remove_request(req);
  270. PR_DestroyLock ( req->req_lock );
  271. req->req_lock = NULL;
  272. slapi_ch_free((void **) &req->req_pblock );
  273. slapi_ch_free((void **) &req );
  274. } else {
  275. return( req->req_tid);
  276. }
  277. }
  278. }
  279. return( NULL);
  280. }
  281. int
  282. sync_persist_startup (PRThread *tid, Sync_Cookie *cookie)
  283. {
  284. SyncRequest *cur;
  285. int rc = 1;
  286. if ( SYNC_IS_INITIALIZED() && NULL != tid ) {
  287. SYNC_LOCK_READ();
  288. /* Find and change */
  289. cur = sync_request_list->sync_req_head;
  290. while ( NULL != cur ) {
  291. if ( cur->req_tid == tid ) {
  292. cur->req_active = PR_TRUE;
  293. cur->req_cookie = cookie;
  294. rc = 0;
  295. break;
  296. }
  297. cur = cur->req_next;
  298. }
  299. SYNC_UNLOCK_READ();
  300. }
  301. return (rc);
  302. }
  303. int
  304. sync_persist_terminate (PRThread *tid)
  305. {
  306. SyncRequest *cur;
  307. int rc = 1;
  308. if ( SYNC_IS_INITIALIZED() && NULL != tid ) {
  309. SYNC_LOCK_READ();
  310. /* Find and change */
  311. cur = sync_request_list->sync_req_head;
  312. while ( NULL != cur ) {
  313. if ( cur->req_tid == tid ) {
  314. cur->req_active = PR_FALSE;
  315. cur->req_complete = PR_TRUE;
  316. rc = 0;
  317. break;
  318. }
  319. cur = cur->req_next;
  320. }
  321. SYNC_UNLOCK_READ();
  322. }
  323. if (rc == 0) {
  324. sync_remove_request(cur);
  325. }
  326. return(rc);
  327. }
  328. int
  329. sync_persist_terminate_all ()
  330. {
  331. SyncRequest *cur;
  332. if ( SYNC_IS_INITIALIZED() ) {
  333. SYNC_LOCK_READ();
  334. cur = sync_request_list->sync_req_head;
  335. while ( NULL != cur ) {
  336. cur->req_complete = PR_TRUE;
  337. cur = cur->req_next;
  338. }
  339. SYNC_UNLOCK_READ();
  340. }
  341. return (0);
  342. }
  343. /*
  344. * Allocate and initialize an empty Sync node.
  345. */
  346. static SyncRequest *
  347. sync_request_alloc()
  348. {
  349. SyncRequest *req;
  350. req = (SyncRequest *) slapi_ch_calloc( 1, sizeof( SyncRequest ));
  351. req->req_pblock = NULL;
  352. if (( req->req_lock = PR_NewLock()) == NULL ) {
  353. slapi_log_error (SLAPI_LOG_FATAL, SYNC_PLUGIN_SUBSYSTEM, "sync_request_alloc: cannot initialize lock structure. ");
  354. slapi_ch_free((void **)&req);
  355. return( NULL );
  356. }
  357. req->req_tid = (PRThread *) NULL;
  358. req->req_complete = 0;
  359. req->req_cookie = NULL;
  360. req->ps_eq_head = req->ps_eq_tail = (SyncQueueNode *) NULL;
  361. req->req_next = NULL;
  362. req->req_active = PR_FALSE;
  363. return req;
  364. }
  365. /*
  366. * Add the given persistent search to the
  367. * head of the list of persistent searches.
  368. */
  369. static int
  370. sync_add_request( SyncRequest *req )
  371. {
  372. int rc = 0;
  373. if ( SYNC_IS_INITIALIZED() && NULL != req ) {
  374. SYNC_LOCK_WRITE();
  375. if (sync_request_list->sync_req_cur_persist < sync_request_list->sync_req_max_persist) {
  376. sync_request_list->sync_req_cur_persist++;
  377. req->req_next = sync_request_list->sync_req_head;
  378. sync_request_list->sync_req_head = req;
  379. } else {
  380. rc = 1;
  381. }
  382. SYNC_UNLOCK_WRITE();
  383. }
  384. return(rc);
  385. }
  386. static void
  387. sync_remove_request( SyncRequest *req )
  388. {
  389. SyncRequest *cur;
  390. int removed = 0;
  391. if ( SYNC_IS_INITIALIZED() && NULL != req ) {
  392. SYNC_LOCK_WRITE();
  393. if ( NULL == sync_request_list->sync_req_head ) {
  394. /* should not happen, attempt to remove a request never added */
  395. } else if ( req == sync_request_list->sync_req_head ) {
  396. /* Remove from head */
  397. sync_request_list->sync_req_head = sync_request_list->sync_req_head->req_next;
  398. removed = 1;
  399. } else {
  400. /* Find and remove from list */
  401. cur = sync_request_list->sync_req_head;
  402. while ( NULL != cur->req_next ) {
  403. if ( cur->req_next == req ) {
  404. cur->req_next = cur->req_next->req_next;
  405. removed = 1;
  406. break;
  407. } else {
  408. cur = cur->req_next;
  409. }
  410. }
  411. }
  412. if (removed) {
  413. sync_request_list->sync_req_cur_persist--;
  414. }
  415. SYNC_UNLOCK_WRITE();
  416. if (!removed) {
  417. slapi_log_error (SLAPI_LOG_PLUGIN, SYNC_PLUGIN_SUBSYSTEM, "attempt to remove nonexistent req");
  418. }
  419. }
  420. }
  421. static void
  422. sync_request_wakeup_all()
  423. {
  424. if ( SYNC_IS_INITIALIZED()) {
  425. PR_Lock( sync_request_list->sync_req_cvarlock );
  426. PR_NotifyAllCondVar( sync_request_list->sync_req_cvar );
  427. PR_Unlock( sync_request_list->sync_req_cvarlock );
  428. }
  429. }
  430. static int
  431. sync_acquire_connection (Slapi_Connection *conn)
  432. {
  433. int rc;
  434. /* need to acquire a reference to this connection so that it will not
  435. be released or cleaned up out from under us
  436. in psearch.c it is implemented as:
  437. PR_Lock( ps->req_pblock->pb_conn->c_mutex );
  438. conn_acq_flag = connection_acquire_nolock(ps->req_pblock->pb_conn);
  439. PR_Unlock( ps->req_pblock->pb_conn->c_mutex );
  440. HOW TO DO FROM A PLUGIN
  441. - either expose the functions from the connection code in the private api
  442. and allow to link them in
  443. - or fake a connection structure
  444. struct fake_conn {
  445. void *needed1
  446. void *needed2
  447. void *pad1
  448. void *pad2
  449. void *needed3;
  450. }
  451. struct fake_conn *c = (struct fake_conn *) conn;
  452. c->needed3 ++;
  453. this would require knowledge or analysis of the connection structure,
  454. could probably be done for servers with a common history
  455. */
  456. /* use exposed slapi_connection functions */
  457. rc = slapi_connection_acquire(conn);
  458. return (rc);
  459. }
  460. static int
  461. sync_release_connection (Slapi_PBlock *pb, Slapi_Connection *conn, Slapi_Operation *op, int release)
  462. {
  463. /* see comments in sync_acquire_connection */
  464. /* using exposed connection handling functions */
  465. slapi_connection_remove_operation(pb, conn, op, release);
  466. return(0);
  467. }
  468. /*
  469. * Thread routine for sending search results to a client
  470. * which is persistently waiting for them.
  471. *
  472. * This routine will terminate when either (a) the ps_complete
  473. * flag is set, or (b) the associated operation is abandoned.
  474. * In any case, the thread won't notice until it wakes from
  475. * sleeping on the ps_list condition variable, so it needs
  476. * to be awakened.
  477. */
  478. static void
  479. sync_send_results( void *arg )
  480. {
  481. SyncRequest *req = (SyncRequest *)arg;
  482. SyncQueueNode *qnode, *qnodenext;
  483. int conn_acq_flag = 0;
  484. Slapi_Connection *conn = NULL;
  485. Slapi_Operation *op = NULL;
  486. int rc;
  487. PRUint64 connid;
  488. int opid;
  489. slapi_pblock_get(req->req_pblock, SLAPI_CONN_ID, &connid);
  490. slapi_pblock_get(req->req_pblock, SLAPI_OPERATION_ID, &opid);
  491. slapi_pblock_get(req->req_pblock, SLAPI_CONNECTION, &conn);
  492. slapi_pblock_get(req->req_pblock, SLAPI_OPERATION, &op);
  493. conn_acq_flag = sync_acquire_connection (conn);
  494. if (conn_acq_flag) {
  495. slapi_log_error(SLAPI_LOG_FATAL, "Content Synchronization Search",
  496. "conn=%" NSPRIu64 " op=%d Could not acquire the connection - aborted\n",
  497. (long long unsigned int)connid, opid);
  498. }
  499. PR_Lock( sync_request_list->sync_req_cvarlock );
  500. while ( (conn_acq_flag == 0) && !req->req_complete ) {
  501. /* Check for an abandoned operation */
  502. Slapi_Operation *op;
  503. slapi_pblock_get(req->req_pblock, SLAPI_OPERATION, &op);
  504. if ( op == NULL || slapi_op_abandoned( req->req_pblock ) ) {
  505. slapi_log_error(SLAPI_LOG_PLUGIN, "Content Synchronization Search",
  506. "conn=%" NSPRIu64 " op=%d Operation no longer active - terminating\n",
  507. (long long unsigned int)connid, opid);
  508. break;
  509. }
  510. if ( NULL == req->ps_eq_head || !req->req_active) {
  511. /* Nothing to do yet, or the refresh phase is not yet completed */
  512. /* If an operation is abandoned, we do not get notified by the
  513. * connection code. Wake up every second to check if thread
  514. * should terminate.
  515. */
  516. PR_WaitCondVar( sync_request_list->sync_req_cvar, PR_SecondsToInterval(1) );
  517. } else {
  518. /* dequeue the item */
  519. int attrsonly;
  520. char **attrs;
  521. char **noattrs = NULL;
  522. LDAPControl **ectrls = NULL;
  523. Slapi_Entry *ec;
  524. int chg_type = LDAP_SYNC_NONE;
  525. /* deque one element */
  526. PR_Lock( req->req_lock );
  527. qnode = req->ps_eq_head;
  528. req->ps_eq_head = qnode->sync_next;
  529. if ( NULL == req->ps_eq_head ) {
  530. req->ps_eq_tail = NULL;
  531. }
  532. PR_Unlock( req->req_lock );
  533. /* Get all the information we need to send the result */
  534. ec = qnode->sync_entry;
  535. slapi_pblock_get( req->req_pblock, SLAPI_SEARCH_ATTRS, &attrs );
  536. slapi_pblock_get( req->req_pblock, SLAPI_SEARCH_ATTRSONLY, &attrsonly );
  537. /*
  538. * Send the result. Since send_ldap_search_entry can block for
  539. * up to 30 minutes, we relinquish all locks before calling it.
  540. */
  541. PR_Unlock(sync_request_list->sync_req_cvarlock);
  542. /*
  543. * The entry is in the right scope and matches the filter
  544. * but we need to redo the filter test here to check access
  545. * controls. See the comments at the slapi_filter_test()
  546. * call in sync_persist_add().
  547. */
  548. if ( slapi_vattr_filter_test( req->req_pblock, ec, req->req_filter,
  549. 1 /* verify_access */ ) == 0 ) {
  550. slapi_pblock_set( req->req_pblock, SLAPI_SEARCH_RESULT_ENTRY, ec );
  551. /* NEED TO BUILD THE CONTROL */
  552. switch (qnode->sync_chgtype){
  553. case LDAP_REQ_ADD:
  554. chg_type = LDAP_SYNC_ADD;
  555. break;
  556. case LDAP_REQ_MODIFY:
  557. chg_type = LDAP_SYNC_MODIFY;
  558. break;
  559. case LDAP_REQ_MODRDN:
  560. chg_type = LDAP_SYNC_MODIFY;
  561. break;
  562. case LDAP_REQ_DELETE:
  563. chg_type = LDAP_SYNC_DELETE;
  564. noattrs = (char **)slapi_ch_calloc(2, sizeof (char *));
  565. noattrs[0] = slapi_ch_strdup("1.1");
  566. noattrs[1] = NULL;
  567. break;
  568. }
  569. ectrls = (LDAPControl **)slapi_ch_calloc(2, sizeof (LDAPControl *));
  570. if (req->req_cookie)
  571. sync_cookie_update(req->req_cookie, ec);
  572. sync_create_state_control(ec, &ectrls[0], chg_type, req->req_cookie);
  573. rc = slapi_send_ldap_search_entry( req->req_pblock,
  574. ec, ectrls,
  575. noattrs?noattrs:attrs, attrsonly );
  576. if (rc) {
  577. slapi_log_error(SLAPI_LOG_CONNS, SYNC_PLUGIN_SUBSYSTEM,
  578. "Error %d sending entry %s\n",
  579. rc, slapi_entry_get_dn_const(ec));
  580. }
  581. ldap_controls_free(ectrls);
  582. slapi_ch_array_free(noattrs);
  583. }
  584. PR_Lock(sync_request_list->sync_req_cvarlock);
  585. /* Deallocate our wrapper for this entry */
  586. sync_node_free( &qnode );
  587. }
  588. }
  589. PR_Unlock( sync_request_list->sync_req_cvarlock );
  590. sync_remove_request( req );
  591. /* indicate the end of search */
  592. sync_release_connection(req->req_pblock, conn, op, conn_acq_flag == 0);
  593. PR_DestroyLock ( req->req_lock );
  594. req->req_lock = NULL;
  595. slapi_ch_free((void **) &req->req_pblock );
  596. slapi_ch_free((void **) &req->req_orig_base );
  597. slapi_filter_free(req->req_filter, 1);
  598. sync_cookie_free(&req->req_cookie);
  599. for ( qnode = req->ps_eq_head; qnode; qnode = qnodenext) {
  600. qnodenext = qnode->sync_next;
  601. sync_node_free( &qnode );
  602. }
  603. slapi_ch_free((void **) &req );
  604. }
  605. /*
  606. * Free a sync update node (and everything it holds).
  607. */
  608. static void
  609. sync_node_free( SyncQueueNode **node )
  610. {
  611. if ( node != NULL && *node != NULL ) {
  612. if ( (*node)->sync_entry != NULL ) {
  613. slapi_entry_free( (*node)->sync_entry );
  614. (*node)->sync_entry = NULL;
  615. }
  616. slapi_ch_free( (void **)node );
  617. }
  618. }