connection_read_operation(Connection *conn, Operation *op, ber_tag_t *tag, int *remaining_data)
{
    ber_len_t len = 0;
    int ret = 0;
    int32_t waits_done = 0;
    ber_int_t msgid;
    int new_operation = 1; /* Are we doing the first I/O read for a new operation ? */
    char *buffer = conn->c_private->c_buffer;
    PRErrorCode err = 0;
    PRInt32 syserr = 0;
    size_t buffer_data_avail;
    int conn_closed = 0;

    pthread_mutex_lock(&(conn->c_mutex));
    /*
     * if the socket is still valid, get the ber element
     * waiting for us on this connection. timeout is handled
     * in the low-level read_function.
     */
    if ((conn->c_sd == SLAPD_INVALID_SOCKET) ||
        (conn->c_flags & CONN_FLAG_CLOSING)) {
        ret = CONN_DONE;
        goto done;
    }

    *tag = LBER_DEFAULT;
    /* First check to see if we have buffered data from "before" */
    if ((buffer_data_avail = conn_buffered_data_avail_nolock(conn, &conn_closed))) {
        /* If so, use that data first */
        if (0 != get_next_from_buffer(buffer + conn->c_private->c_buffer_offset,
                                      buffer_data_avail,
                                      &len, tag, op->o_ber, conn)) {
            ret = CONN_DONE;
            goto done;
        }
        new_operation = 0;
    }
    /* If we still haven't seen a complete PDU, read from the network */
    while (*tag == LBER_DEFAULT) {
        int32_t ioblocktimeout_waits = conn->c_ioblocktimeout / CONN_TURBO_TIMEOUT_INTERVAL;
        /* We should never get here with data remaining in the buffer */
        PR_ASSERT(!new_operation || !conn_buffered_data_avail_nolock(conn, &conn_closed));
        /* We make a non-blocking read call */
        if (CONNECTION_BUFFER_OFF != conn->c_private->use_buffer) {
            ret = connection_read_ldap_data(conn, &err);
        } else {
            ret = get_next_from_buffer(NULL, 0, &len, tag, op->o_ber, conn);
            if (ret == -1) {
                ret = CONN_DONE;
                goto done; /* get_next_from_buffer does the disconnect stuff */
            } else if (ret == 0) {
                ret = len;
            }
            *remaining_data = 0;
        }
        if (ret <= 0) {
            if (0 == ret) {
                /* Connection is closed */
                disconnect_server_nomutex(conn, conn->c_connid, -1, SLAPD_DISCONNECT_BAD_BER_TAG, 0);
                conn->c_gettingber = 0;
                signal_listner();
                ret = CONN_DONE;
                goto done;
            }
            /* err = PR_GetError(); */
            /* If we would block, we need to poll for a while */
            syserr = PR_GetOSError();
            if (SLAPD_PR_WOULD_BLOCK_ERROR(err) ||
                SLAPD_SYSTEM_WOULD_BLOCK_ERROR(syserr)) {
                struct PRPollDesc pr_pd;
                PRIntervalTime timeout = PR_MillisecondsToInterval(CONN_TURBO_TIMEOUT_INTERVAL);
                pr_pd.fd = (PRFileDesc *)conn->c_prfd;
                pr_pd.in_flags = PR_POLL_READ;
                pr_pd.out_flags = 0;
                PR_Lock(conn->c_pdumutex);
                ret = PR_Poll(&pr_pd, 1, timeout);
                PR_Unlock(conn->c_pdumutex);
                waits_done++;
                /* Did we time out ? */
                if (0 == ret) {
                    /* We timed out, should the server shutdown ? */
                    if (op_shutdown) {
                        ret = CONN_SHUTDOWN;
                        goto done;
                    }
                    /* We timed out, is this the first read in a PDU ? */
                    if (new_operation) {
                        /* If so, we return */
                        ret = CONN_TIMEDOUT;
                        goto done;
                    } else {
                        /* Otherwise we loop, unless we exceeded the ioblock timeout */
                        if (waits_done > ioblocktimeout_waits) {
                            slapi_log_err(SLAPI_LOG_CONNS, "connection_read_operation",
                                          "ioblocktimeout expired on connection %" PRIu64 "\n", conn->c_connid);
                            disconnect_server_nomutex(conn, conn->c_connid, -1,
                                                      SLAPD_DISCONNECT_IO_TIMEOUT, 0);
                            ret = CONN_DONE;
                            goto done;
                        } else {

                            /* The turbo mode may cause threads starvation.
                                  Do a yield here to reduce the starving.
                            */
                            PR_Sleep(PR_INTERVAL_NO_WAIT);

                            continue;
                        }
                    }
                }
                if (-1 == ret) {
                    /* PR_Poll call failed */
                    err = PR_GetError();
                    syserr = PR_GetOSError();
                    slapi_log_err(SLAPI_LOG_ERR, "connection_read_operation",
                                  "PR_Poll for connection %" PRIu64 " returns %d (%s)\n", conn->c_connid, err, slapd_pr_strerror(err));
                    /* If this happens we should close the connection */
                    disconnect_server_nomutex(conn, conn->c_connid, -1, err, syserr);
                    ret = CONN_DONE;
                    goto done;
                }
                slapi_log_err(SLAPI_LOG_CONNS,
                              "connection_read_operation", "connection %" PRIu64 " waited %d times for read to be ready\n", conn->c_connid, waits_done);
            } else {
                /* Some other error, typically meaning bad stuff */
                syserr = PR_GetOSError();
                slapi_log_err(SLAPI_LOG_CONNS, "connection_read_operation",
                              "PR_Recv for connection %" PRIu64 " returns %d (%s)\n", conn->c_connid, err, slapd_pr_strerror(err));
                /* If this happens we should close the connection */
                disconnect_server_nomutex(conn, conn->c_connid, -1, err, syserr);
                ret = CONN_DONE;
                goto done;
            }
        } else {
            /* We read some data off the network, do something with it */
            if (CONNECTION_BUFFER_OFF != conn->c_private->use_buffer) {
                conn->c_private->c_buffer_bytes = ret;
                conn->c_private->c_buffer_offset = 0;

                if (get_next_from_buffer(buffer,
                                         conn->c_private->c_buffer_bytes - conn->c_private->c_buffer_offset,
                                         &len, tag, op->o_ber, conn) != 0) {
                    ret = CONN_DONE;
                    goto done;
                }
            }
            slapi_log_err(SLAPI_LOG_CONNS,
                          "connection_read_operation", "connection %" PRIu64 " read %d bytes\n", conn->c_connid, ret);

            new_operation = 0;
            ret = CONN_FOUND_WORK_TO_DO;
            waits_done = 0; /* got some data: reset counter */
        }
    }
    /* If there is remaining buffered data, set the flag to tell the caller */
    if (conn_buffered_data_avail_nolock(conn, &conn_closed)) {
        *remaining_data = 1;
    } else if (conn_closed) {
        /* connection closed */
        ret = CONN_DONE;
        goto done;
    }

    if (*tag != LDAP_TAG_MESSAGE) {
        /*
         * We received a non-LDAP message.  Log and close connection.
         */
        slapi_log_err(SLAPI_LOG_ERR,
                      "connection_read_operation", "conn=%" PRIu64 " received a non-LDAP message (tag 0x%lx, expected 0x%lx)\n",
                      conn->c_connid, *tag, LDAP_TAG_MESSAGE);
        disconnect_server_nomutex(conn, conn->c_connid, -1,
                                  SLAPD_DISCONNECT_BAD_BER_TAG, EPROTO);
        ret = CONN_DONE;
        goto done;
    }

    if ((*tag = ber_get_int(op->o_ber, &msgid)) != LDAP_TAG_MSGID) {
        /* log, close and send error */
        slapi_log_err(SLAPI_LOG_ERR,
                      "connection_read_operation", "conn=%" PRIu64 " unable to read tag for incoming request\n", conn->c_connid);
        disconnect_server_nomutex(conn, conn->c_connid, -1, SLAPD_DISCONNECT_BAD_BER_TAG, EPROTO);
        ret = CONN_DONE;
        goto done;
    }
    if (is_ber_too_big(conn, len)) {
        disconnect_server_nomutex(conn, conn->c_connid, -1, SLAPD_DISCONNECT_BER_TOO_BIG, 0);
        ret = CONN_DONE;
        goto done;
    }
    op->o_msgid = msgid;

    *tag = ber_peek_tag(op->o_ber, &len);
    switch (*tag) {
    case LBER_ERROR:
    case LDAP_TAG_LDAPDN: /* optional username, for CLDAP */
        /* log, close and send error */
        slapi_log_err(SLAPI_LOG_ERR,
                      "connection_read_operation", "conn=%" PRIu64 " ber_peek_tag returns 0x%lx\n", conn->c_connid, *tag);
        disconnect_server_nomutex(conn, conn->c_connid, -1, SLAPD_DISCONNECT_BER_PEEK, EPROTO);
        ret = CONN_DONE;
        goto done;
    default:
        break;
    }
    op->o_tag = *tag;
done:
    pthread_mutex_unlock(&(conn->c_mutex));
    return ret;
}