public void run() {
            synchronized (stateMonitor) {
                try {
                    scheduledSendCreateVisitors = false;
                    if (done) {
                        return; // Session already closed; we must not touch anything else.
                    }
                    // We both send requests and process replies in the context of a dedicated task executor pool.
                    // However, MessageBus sending and reply receiving happens in the context of entirely
                    // separate threads. If the backend responds very quickly to visitor requests (such as
                    // if buckets are empty), this can leave us in the following awkward position:
                    //
                    //   1. Replies arrive from backend, open up the throttle window, reply handling
                    //      task gets pushed onto executor queue (but not yet executed).
                    //   2. Send loop below continuously get a free send slot, keeps sending visitors
                    //      and filling up the set of pending buckets in the progress token.
                    //   3. Since visitor session is busy-looping in the send task, reply processing is
                    //      consequently entirely starved until the MessageBus throttle window is bursting
                    //      at the seams. This can effectively nullify the effects of the throttling policy,
                    //      especially if it's dynamic. But a static throttle policy with a sufficiently
                    //      high max window size will also potentially cause a runaway visitor train since
                    //      the active window size keeps getting decreased by backend replies.
                    //
                    // To get around this, we explicitly check for concurrently scheduled message handling
                    // tasks from the transport layer, breaking the loop if at least one handler has been
                    // scheduled. This also has the (positive) effect of draining all reply tasks before we
                    // start sending more work downstream.
                    //
                    // Since visitor session progress is edge-triggered and progresses exclusively by sending
                    // new visitors in reply handling tasks, it's critical that we never end up in a situation
                    // where we have no pending CreateVisitors (or scheduled tasks), or we risk effectively
                    // hanging the session. We must therefore be very careful that we only exit the send loop
                    // if we _know_ we have at least one pending task enqueued that will ensure session progress.
                    //
                    // We're holding the session (token) lock around checking the pending reply tasks count, so
                    // if we observe a change we know that a reply task must have been scheduled and that its
                    // processing must take place sequenced after we have exited the loop, as the reply handling
                    // also takes the session (token) lock. I.e. it should not be possible to end up in a
                    // situation where we stall session progress due to not having any further event edges.
                    while (progress.getIterator().hasNext() && !hasScheduledHandleReplyTask()) {
                        VisitorIterator.BucketProgress bucket = progress.getIterator().getNext();
                        Result result = sender.send(createMessage(bucket));
                        if (result.isAccepted()) {
                            log.log(Level.FINE, () -> sessionName + ": sent CreateVisitor for bucket " +
                                    bucket.getSuperbucket() + " with progress " + bucket.getProgress());
                            ++pendingMessageCount;
                        } else {
                            // Must reinsert bucket without progress into iterator since
                            // we failed to send visitor.
                            progress.getIterator().update(bucket.getSuperbucket(), bucket.getProgress());
                            break;
                        }
                    }
                } catch (Exception e) {
                    String msg = "Got exception of type " + e.getClass().getName() +
                            " with message '" + e.getMessage() +
                            "' while attempting to send visitors";
                    log.log(Level.WARNING, msg);
                    transitionTo(new StateDescription(State.FAILED, msg));
                    // It's likely that the exception caused a failure to send a
                    // visitor message, meaning we won't get a reply task in the
                    // future from which we can execute logic to complete the
                    // session. Thusly, we have to do this here and now.
                    continueVisiting();
                } catch (Throwable t) {
                    // We can't reliably handle this; take a nosedive
                    com.yahoo.protect.Process.logAndDie("Caught unhandled error when trying to send visitors", t);
                }
            }
        }
Called from the constructor to ensure control and data handlers
are set and initialized.
documentapi/src/main/java/com/yahoo/documentapi/messagebus/MessageBusVisitorSession.java