diff options
Diffstat (limited to 'src/or/relay.c')
| -rw-r--r-- | src/or/relay.c | 114 |
1 files changed, 107 insertions, 7 deletions
diff --git a/src/or/relay.c b/src/or/relay.c index 3419e3d19..c81b8311a 100644 --- a/src/or/relay.c +++ b/src/or/relay.c @@ -533,6 +533,12 @@ relay_send_command_from_edge(uint16_t stream_id, circuit_t *circ, log_debug(LD_OR,"delivering %d cell %s.", relay_command, cell_direction == CELL_DIRECTION_OUT ? "forward" : "backward"); + /* If we are sending an END cell and this circuit is used for a tunneled + * directory request, advance its state. */ + if (relay_command == RELAY_COMMAND_END && circ->dirreq_id) + geoip_change_dirreq_state(circ->dirreq_id, DIRREQ_TUNNELED, + DIRREQ_END_CELL_SENT); + if (cell_direction == CELL_DIRECTION_OUT && circ->n_conn) { /* if we're using relaybandwidthrate, this conn wants priority */ circ->n_conn->client_used = approx_time(); @@ -999,7 +1005,8 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, relay_header_unpack(&rh, cell->payload); // log_fn(LOG_DEBUG,"command %d stream %d", rh.command, rh.stream_id); num_seen++; - log_debug(domain, "Now seen %d relay cells here.", num_seen); + log_debug(domain, "Now seen %d relay cells here (command %d, stream %d).", + num_seen, rh.command, rh.stream_id); if (rh.length > RELAY_PAYLOAD_SIZE) { log_fn(LOG_PROTOCOL_WARN, LD_PROTOCOL, @@ -1038,6 +1045,16 @@ connection_edge_process_relay_cell(cell_t *cell, circuit_t *circ, "Begin cell for known stream. Dropping."); return 0; } + if (rh.command == RELAY_COMMAND_BEGIN_DIR) { + /* Assign this circuit and its app-ward OR connection a unique ID, + * so that we can measure download times. The local edge and dir + * connection will be assigned the same ID when they are created + * and linked. */ + static uint64_t next_id = 0; + circ->dirreq_id = ++next_id; + TO_CONN(TO_OR_CIRCUIT(circ)->p_conn)->dirreq_id = circ->dirreq_id; + } + return connection_exit_begin_conn(cell, circ); case RELAY_COMMAND_DATA: ++stats_n_data_cells_received; @@ -1358,7 +1375,7 @@ connection_edge_consider_sending_sendme(edge_connection_t *conn) return; } - while (conn->deliver_window < STREAMWINDOW_START - STREAMWINDOW_INCREMENT) { + while (conn->deliver_window <= STREAMWINDOW_START - STREAMWINDOW_INCREMENT) { log_debug(conn->cpath_layer?LD_APP:LD_EXIT, "Outbuf %d, Queuing stream sendme.", (int)conn->_base.outbuf_flushlen); @@ -1472,7 +1489,7 @@ circuit_consider_sending_sendme(circuit_t *circ, crypt_path_t *layer_hint) { // log_fn(LOG_INFO,"Considering: layer_hint is %s", // layer_hint ? "defined" : "null"); - while ((layer_hint ? layer_hint->deliver_window : circ->deliver_window) < + while ((layer_hint ? layer_hint->deliver_window : circ->deliver_window) <= CIRCWINDOW_START - CIRCWINDOW_INCREMENT) { log_debug(LD_CIRC,"Queuing circuit sendme."); if (layer_hint) @@ -1508,6 +1525,10 @@ static int total_cells_allocated = 0; /** A memory pool to allocate packed_cell_t objects. */ static mp_pool_t *cell_pool = NULL; +/** Memory pool to allocate insertion_time_elem_t objects used for cell + * statistics. */ +static mp_pool_t *it_pool = NULL; + /** Allocate structures to hold cells. */ void init_cell_pool(void) @@ -1516,7 +1537,8 @@ init_cell_pool(void) cell_pool = mp_pool_new(sizeof(packed_cell_t), 128*1024); } -/** Free all storage used to hold cells. */ +/** Free all storage used to hold cells (and insertion times if we measure + * cell statistics). */ void free_cell_pool(void) { @@ -1525,6 +1547,10 @@ free_cell_pool(void) mp_pool_destroy(cell_pool); cell_pool = NULL; } + if (it_pool) { + mp_pool_destroy(it_pool); + it_pool = NULL; + } } /** Free excess storage in cell pool. */ @@ -1599,7 +1625,37 @@ cell_queue_append(cell_queue_t *queue, packed_cell_t *cell) void cell_queue_append_packed_copy(cell_queue_t *queue, const cell_t *cell) { - cell_queue_append(queue, packed_cell_copy(cell)); + packed_cell_t *copy = packed_cell_copy(cell); + /* Remember the time when this cell was put in the queue. */ + if (get_options()->CellStatistics) { + struct timeval now; + uint32_t added; + insertion_time_queue_t *it_queue = queue->insertion_times; + if (!it_pool) + it_pool = mp_pool_new(sizeof(insertion_time_elem_t), 1024); + tor_gettimeofday(&now); +#define SECONDS_IN_A_DAY 86400L + added = (now.tv_sec % SECONDS_IN_A_DAY) * 100L + now.tv_usec / 10000L; + if (!it_queue) { + it_queue = tor_malloc_zero(sizeof(insertion_time_queue_t)); + queue->insertion_times = it_queue; + } + if (it_queue->last && it_queue->last->insertion_time == added) { + it_queue->last->counter++; + } else { + insertion_time_elem_t *elem = mp_pool_get(it_pool); + elem->next = NULL; + elem->insertion_time = added; + elem->counter = 1; + if (it_queue->last) { + it_queue->last->next = elem; + it_queue->last = elem; + } else { + it_queue->first = it_queue->last = elem; + } + } + } + cell_queue_append(queue, copy); } /** Remove and free every cell in <b>queue</b>. */ @@ -1615,6 +1671,14 @@ cell_queue_clear(cell_queue_t *queue) } queue->head = queue->tail = NULL; queue->n = 0; + if (queue->insertion_times) { + while (queue->insertion_times->first) { + insertion_time_elem_t *elem = queue->insertion_times->first; + queue->insertion_times->first = elem->next; + mp_pool_release(elem); + } + tor_free(queue->insertion_times); + } } /** Extract and return the cell at the head of <b>queue</b>; return NULL if @@ -1667,7 +1731,7 @@ prev_circ_on_conn_p(circuit_t *circ, or_connection_t *conn) } /** Add <b>circ</b> to the list of circuits with pending cells on - * <b>conn</b>. No effect if <b>circ</b> is already unlinked. */ + * <b>conn</b>. No effect if <b>circ</b> is already linked. */ void make_circuit_active_on_conn(circuit_t *circ, or_connection_t *conn) { @@ -1693,7 +1757,7 @@ make_circuit_active_on_conn(circuit_t *circ, or_connection_t *conn) assert_active_circuits_ok_paranoid(conn); } -/** Remove <b>circ</b> to the list of circuits with pending cells on +/** Remove <b>circ</b> from the list of circuits with pending cells on * <b>conn</b>. No effect if <b>circ</b> is already unlinked. */ void make_circuit_inactive_on_conn(circuit_t *circ, or_connection_t *conn) @@ -1808,6 +1872,42 @@ connection_or_flush_from_first_active_circuit(or_connection_t *conn, int max, packed_cell_t *cell = cell_queue_pop(queue); tor_assert(*next_circ_on_conn_p(circ,conn)); + /* Calculate the exact time that this cell has spent in the queue. */ + if (get_options()->CellStatistics && !CIRCUIT_IS_ORIGIN(circ)) { + struct timeval now; + uint32_t flushed; + uint32_t cell_waiting_time; + insertion_time_queue_t *it_queue = queue->insertion_times; + tor_gettimeofday(&now); + flushed = (now.tv_sec % SECONDS_IN_A_DAY) * 100L + + now.tv_usec / 10000L; + if (!it_queue || !it_queue->first) { + log_warn(LD_BUG, "Cannot determine insertion time of cell."); + } else { + or_circuit_t *orcirc = TO_OR_CIRCUIT(circ); + insertion_time_elem_t *elem = it_queue->first; + cell_waiting_time = (flushed * 10L + SECONDS_IN_A_DAY * 1000L - + elem->insertion_time * 10L) % (SECONDS_IN_A_DAY * 1000L); +#undef SECONDS_IN_A_DAY + elem->counter--; + if (elem->counter < 1) { + it_queue->first = elem->next; + if (elem == it_queue->last) + it_queue->last = NULL; + mp_pool_release(elem); + } + orcirc->total_cell_waiting_time += cell_waiting_time; + orcirc->processed_cells++; + } + } + + /* If we just flushed our queue and this circuit is used for a + * tunneled directory request, possibly advance its state. */ + if (queue->n == 0 && TO_CONN(conn)->dirreq_id) + geoip_change_dirreq_state(TO_CONN(conn)->dirreq_id, + DIRREQ_TUNNELED, + DIRREQ_CIRC_QUEUE_FLUSHED); + connection_write_to_buf(cell->body, CELL_NETWORK_SIZE, TO_CONN(conn)); packed_cell_free(cell); |