+  * Copyright (C) 2018 William Brown <william@blackhats.net.au>

+ static void add_work_q(work_q_item *, Operation *);

+ static work_q_item *get_work_q(Operation **);

+     Operation *op;

+         (*work_q)->op = NULL;

+ static Operation *

+     return operation_new(plugin_build_operation_action_bitmap(0, plugin_get_server_plg()));

+ 

+ connection_done_operation(Connection *conn, Operation *op)

+     operation_free(&op, conn);

+      *

+      * It's happening!!!!

+      * William Sep 2018

+     slapi_log_err(SLAPI_LOG_CONNS, "connection_reset", "new %sconnection on fd=%d\n", pTmp, conn->c_sd);

+     for (size_t i = 0; i < max_threads; i++) {

+             /* ns_connection_post_io_or_closing(conn); */

+             ns_handle_closure_conn_nomutex(conn);

+ connection_acquire_nolock(Connection *conn)

+     if (conn->c_flags & CONN_FLAG_CLOSING) {

+         slapi_log_err(SLAPI_LOG_ERR, "connection_acquire_nolock",

+     Operation *op = NULL;

+     op = connection_get_operation();

+     slapi_pblock_set(pb, SLAPI_OPERATION, op);

+     connection_add_operation(conn, op);

+     Operation *op = NULL;

+     } else if (NULL == (wqitem = get_work_q(&op))) {

+         slapi_pblock_set(pb, SLAPI_OPERATION, op);

+ connection_read_operation_nolock(Connection *conn, Operation *op, ber_tag_t *tag, int *remaining_data)

+         return CONN_DONE;

+             return CONN_DONE;

+                 return CONN_DONE;

+                 return CONN_DONE;

+             if (SLAPD_PR_WOULD_BLOCK_ERROR(err) || SLAPD_SYSTEM_WOULD_BLOCK_ERROR(syserr)) {

+                         return CONN_SHUTDOWN;

+                         return CONN_TIMEDOUT;

+                             return CONN_DONE;

+                     return CONN_DONE;

+                 return CONN_DONE;

+                     return CONN_DONE;

+         return CONN_DONE;

+         return CONN_DONE;

+         return CONN_DONE;

+         return CONN_DONE;

+         return CONN_DONE;

+ int

+ connection_read_operation(Connection *conn, Operation *op, ber_tag_t *tag, int *remaining_data) {

+     int ret = 0;

+     ret = connection_read_operation_nolock(conn, op, tag, remaining_data);

+     return ret;

+ static void

+     slapi_log_err(SLAPI_LOG_CONNS, "connection_make_readable_nolock",

+         "making readable conn %" PRIu64 " fd=%d\n",

+         conn->c_connid, conn->c_sd);

+     signal_listner();

+              * can't free it or init it here - just set operation to NULL.

+              * ps_send_results will call connection_remove_operation_ext to free it

+              */

+                 /*

+                  * If is_timedout is 1, then thread_turbo_flag is set to 0, so we

+                  * move the check for timeout to the outer and then continue. This resolves

+                  * a coverity issue for dead code detection.

+                  */

+                 if (replication_connection || (is_timedout == 1)) {

+                     connection_make_readable_nolock(conn);

+                     need_wakeup = 1;

+                 }

+ 

+ void

+ ns_connection_cleanup_nolock(Connection *c, Operation *op, Slapi_PBlock *pb) {

+     /* Destroy the pblock (with conditions if psearch) */

+     if (op->o_flags & OP_FLAG_PS) {

+         /*

+          * If this operation begins a psearch, then we detach the operation

+          * from the pblock. When we call pb_destroy, this prevents the op being

+          * destroyed. If we don't we trigger use-after-free!

+          *

+          * ps_send_results will call connection_remove_operation_ext to free it

+          * later for us.

+          */

+         slapi_pblock_set(pb, SLAPI_OPERATION, NULL);

+     } else {

+         connection_remove_operation_ext(pb, c, op);

+     }

+ 

+     /* Remove the refcnt from ns_connection_activity */

+     connection_release_nolock_ext(c, 1);

+     /* Finally kill the pb */

+     slapi_pblock_destroy(pb);

+ }

+ 

+ void

+ ns_connection_do_work(Connection *conn, Operation *op, Slapi_PBlock *pb) {

+ 

+     /* are we in referral-only mode? */

+     if (config_check_referral_mode() && op->o_tag != LDAP_REQ_UNBIND) {

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_do_work",

+                       "conn %" PRIu64 " for fd=%d - sending referral\n",

+                       conn->c_connid, conn->c_sd);

+         referral_mode_reply(pb);

+     } else if (connection_need_new_password(conn, op, pb)) {

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_do_work",

+                       "conn %" PRIu64 " for fd=%d - new password required\n",

+                       conn->c_connid, conn->c_sd);

+ 

+         /* check if new password is required */

+         return;

+     } else if (conn->c_flags & CONN_FLAG_IMPORT) {

+         /* if this is a bulk import, only "add" and "import done"

+          * are allowed */

+         if ((op->o_tag != LDAP_REQ_ADD) && (op->o_tag != LDAP_REQ_EXTENDED)) {

+             /* no cookie for you. */

+             slapi_log_err(SLAPI_LOG_ERR, "ns_connection_do_work",

+                           "conn %" PRIu64 " for fd=%d - attempted operation %lu from within bulk import\n",

+                           conn->c_connid, conn->c_sd, op->o_tag);

+             slapi_send_ldap_result(pb, LDAP_PROTOCOL_ERROR, NULL, NULL, 0, NULL);

+         }

+     } else {

+         /*

+          * Call the do_<operation> function to process this request.

+          */

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_do_work",

+                       "conn %" PRIu64 " for fd=%d - connection_dispatch_operation\n",

+                       conn->c_connid, conn->c_sd);

+         connection_dispatch_operation(conn, op, pb);

+     }

+ }

+ 

+ void

+ ns_handle_work_done(struct ns_job_t *job) {

+     Slapi_PBlock *pb = (Slapi_PBlock *)ns_job_get_data(job);

+     PR_ASSERT(pb);

+     Connection *c = NULL;

+     Operation *op = NULL;

+     slapi_pblock_get(pb, SLAPI_CONNECTION, &c);

+     slapi_pblock_get(pb, SLAPI_OPERATION, &op);

+     PR_ASSERT(c);

+     PR_ASSERT(op);

+ 

+     pthread_mutex_lock(&(c->c_mutex));

+     slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_work_done",

+                   "conn %" PRIu64 " for fd=%d - removing async operation references\n",

+                   c->c_connid, c->c_sd);

+     ns_connection_cleanup_nolock(c, op, pb);

+     pthread_mutex_unlock(&(c->c_mutex));

+ }

+ 

+ void

+ ns_handle_work(struct ns_job_t *job) {

+     /* Actually do something with the operation! */

+     /* All the IO should be done, just do the op, then write the result. */

+     Slapi_PBlock *pb = (Slapi_PBlock *)ns_job_get_data(job);

+     PR_ASSERT(pb);

+     Connection *c = NULL;

+     Operation *op = NULL;

+     slapi_pblock_get(pb, SLAPI_CONNECTION, &c);

+     slapi_pblock_get(pb, SLAPI_OPERATION, &op);

+     PR_ASSERT(c);

+     PR_ASSERT(op);

+     /*

+      * Do we have the resources for the operation?

+      */

+ 

+     pthread_mutex_lock(&(c->c_mutex));

+     int32_t maxthreads = config_get_maxthreadsperconn();

+     if (c->c_threadnumber >= maxthreads) {

+         c->c_maxthreadsblocked++;

+         /* Requeue for a future attempt */

+         ns_job_rearm(job);

+         slapi_log_err(SLAPI_LOG_WARNING, "ns_handle_work",

+                       "conn %" PRIu64 " for fd=%d - maxthreads %"PRId32" of %"PRId32" per conn limit reached, requeuing\n",

+                       c->c_connid, c->c_sd, c->c_threadnumber, maxthreads);

+         pthread_mutex_unlock(&(c->c_mutex));

+         return;

+     } else {

+         c->c_threadnumber += 1;

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_work",

+                       "conn %" PRIu64 " for fd=%d - maxthreads %"PRId32" of %"PRId32" per conn, starting operation ...\n",

+                       c->c_connid, c->c_sd, c->c_threadnumber, maxthreads);

+ 

+         /* Some stolen code for stats of thread counting */

+         if (c->c_threadnumber == maxthreads) {

+             c->c_maxthreadscount++;

+             slapi_counter_increment(max_threads_count);

+             slapi_counter_increment(conns_in_maxthreads);

+             slapi_counter_increment(g_get_global_snmp_vars()->ops_tbl.dsConnectionsInMaxThreads);

+         }

+ 

+         pthread_mutex_unlock(&(c->c_mutex));

+     }

+ 

+     /*

+      ********************

+      * DO THE OPERATION *

+      ********************

+      */

+     slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_work",

+                   "conn %" PRIu64 " for fd=%d - started operation\n",

+                   c->c_connid, c->c_sd);

+     ns_connection_do_work(c, op, pb);

+     slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_work",

+                   "conn %" PRIu64 " for fd=%d - ended operation\n",

+                   c->c_connid, c->c_sd);

+ 

+     /*

+      * WARNING!!! It is UNSAFE to call the direct ns close functions

+      * for the connection here. Use disconnect_server_nomutex instead.

+      */

+ 

+     /* The thread is complete */

+     pthread_mutex_lock(&(c->c_mutex));

+     c->c_threadnumber -= 1;

+     slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_work",

+                   "conn %" PRIu64 " for fd=%d - maxthreads %"PRId32" of %"PRId32" per conn, at end of operation\n",

+                   c->c_connid, c->c_sd, c->c_threadnumber, maxthreads);

+     pthread_mutex_unlock(&(c->c_mutex));

+ 

+     /* Destroy this worker at last - this frees everything! */

+     ns_job_done(job);

+ }

+ 

+ /*

+  * The caller here still has c mutex held.

+  */

+ void

+ ns_handle_add_work(Connection *c, Slapi_PBlock *pb) {

+     /* Add a job to async handle the operation itself. */

+     ns_result_t job_result = ns_add_job(c->c_tp, NS_JOB_THREAD, ns_handle_work, ns_handle_work_done, pb, NULL);

+ 

+     if (job_result != NS_SUCCESS) {

+         if (job_result == NS_SHUTDOWN) {

+             slapi_log_err(SLAPI_LOG_ERR, "ns_handle_add_work",

+                           "conn %" PRIu64 " for fd=%d - async operation failed to be added to queue as server is shuttdng down\n",

+                           c->c_connid, c->c_sd);

+         } else {

+             slapi_log_err(SLAPI_LOG_ERR, "ns_handle_add_work",

+                           "conn %" PRIu64 " for fd=%d - async operation failed to be added to queue error %d\n",

+                           c->c_connid, c->c_sd, job_result);

+         }

+     } else {

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_handle_add_work",

+                       "conn %" PRIu64 " for fd=%d - async operation added to work queue\n",

+                       c->c_connid, c->c_sd);

+     }

+ }

+ 

+ static Operation *

+ connection_activity_ext(Connection *conn, int maxthreads) {

+     Operation *op = NULL;

+         return NULL;

+     /*

+      * With nunc stans we count threads differently. As a result

+      * If we see maxthreads == 0 here, we skip the ++. Remove this hack job

+      * when we gut the old conn code.

+      */

+     if (maxthreads > 0) {

+         conn->c_threadnumber++;

+     }

+ 

+     op = connection_get_operation();

+     connection_add_operation(conn, op);

+     return op;

+ }

+ 

+ /*

+  * Called as from the ns_handle_pr_read_ready callback. That function

+  * holds the NS job lock then the conn lock, so we can just use nomutex

+  * functions.

+  *

+  * This guarantees we are the only IO on the conn at the time.

+  *

+  * It also helpfully seperates an IO event thread from a worker doing

+  * a search. This just sets up the worker ready to go. This means most

+  * connections can have 1:N IO:Workers in operation, without rely on

+  * weird locking tricks.

+  */

+ int32_t

+ ns_connection_activity(Connection *conn, int32_t maxthreads) {

+     /* THIS ACQUIRES THE CONNECTION */

+     /* We added one to the refcnt!!! Remember to remove it! */

+     Operation *op = connection_activity_ext(conn, maxthreads);

+     if (op == NULL) {

+         return -1;

+     }

+ 

+     ber_tag_t tag = 0;

+     int32_t more_data = 0;

+     int32_t result = 0;

+     /* Actually read a PDU here and get it ready .... */

+     int cret = connection_read_operation_nolock(conn, op, &tag, &more_data);

+     slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_activity",

+                   "conn %" PRIu64 " - cret %d\n", conn->c_connid, cret);

+     switch (cret) {

+     case CONN_FOUND_WORK_TO_DO:

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_activity",

+                       "conn %" PRIu64 " read operation successfully - more_data %d "

+                       "ops_initiated %d refcnt %d flags %d\n",

+                       conn->c_connid, more_data,

+                       conn->c_opsinitiated, conn->c_refcnt, conn->c_flags);

+         /* Create the pb and set items into it. */

+         Slapi_PBlock *pb = slapi_pblock_new();

+         slapi_pblock_set(pb, SLAPI_CONNECTION, conn);

+         slapi_pblock_set(pb, SLAPI_OPERATION, op);

+         /* IF REPLICATION || UNBIND RUN IT NOW IN THIS THREAD */

+         if (tag == LDAP_REQ_UNBIND || conn->c_isreplication_session) {

+             /*

+              * This won't deadlock anything because this won't spawn

+              * new threads, and we hold job lock AND the conn lock.

+              * As both are monitors, we are pretty safe here.

+              */

+             ns_connection_do_work(conn, op, pb);

+             /* Now clean up after ourselves. */

+             ns_connection_cleanup_nolock(conn, op, pb);

+             /* Tell our caller not to rearm after an unbind */

+             if (tag == LDAP_REQ_UNBIND) {

+                 result = -4;

+             }

+ 

+         } else {

+             /*

+              * schedule for async dispatch instead.

+              * We don't need to worry about max threads here, because the

+              * worker will just reset the job if it's too busy.

+              */

+             ns_handle_add_work(conn, pb);

+         }

+         break;

+     case CONN_DONE:

+     case CONN_TIMEDOUT:

+         slapi_log_err(SLAPI_LOG_CONNS, "ns_connection_activity",

+                       "conn %" PRIu64 " read not ready due to %d - more_data %d "

+                       "ops_initiated %d refcnt %d flags %d\n",

+                       conn->c_connid, cret, more_data,

+                       conn->c_opsinitiated, conn->c_refcnt, conn->c_flags);

+         connection_release_nolock_ext(conn, 1);

+         /* Free the allocated operation. */

+         connection_remove_operation(conn, op);

+         connection_done_operation(conn, op);

+         result = -2;

+         break;

+     default:

+         slapi_log_err(SLAPI_LOG_ERR, "ns_connection_activity",

+                       "conn %" PRIu64 " read operation failed UNKNOWN %d - more_data %d "

+                       "ops_initiated %d refcnt %d flags %d\n",

+                       conn->c_connid, cret, more_data,

+                       conn->c_opsinitiated, conn->c_refcnt, conn->c_flags);

+         connection_release_nolock_ext(conn, 1);

+         /* Free the allocated operation. */

+         connection_remove_operation(conn, op);

+         connection_done_operation(conn, op);

+         result = -3;

+         break;

+     }

+ 

+     return result;

+ }

+ 

+ int32_t

+ connection_activity(Connection *conn, int maxthreads)

+ {

+     Operation *op = connection_activity_ext(conn, maxthreads);

+     /* Add conn to the end of the work queue.  */

+     /* have to do this last - add_work_q will signal waiters in connection_wait_for_new_work */

+     if (op != NULL) {

+         add_work_q((work_q_item *)conn, op);

+         return 0;

+     }

+     return -1;

+ add_work_q(work_q_item *wqitem, Operation *op)

+     new_work_q->op = op;

+ get_work_q(Operation **op)

+     *op = tmp->op;

+                   "slapd shutting down - signaling operation threads - max work q size %d max work q stack size %d\n",