Remember optimistically sent data until we have gotten a CONNECTED

Since we can retry failed streams under some circumstances, we need
to be ready to send data queued on them.
This commit is contained in:
Nick Mathewson 2011-07-18 15:38:05 -04:00
parent 34a52534bb
commit 218e84b634
4 changed files with 68 additions and 7 deletions

View File

@ -444,7 +444,12 @@ _connection_free(connection_t *conn)
tor_free(edge_conn->chosen_exit_name); tor_free(edge_conn->chosen_exit_name);
if (edge_conn->socks_request) if (edge_conn->socks_request)
socks_request_free(edge_conn->socks_request); socks_request_free(edge_conn->socks_request);
if (edge_conn->pending_optimistic_data) {
generic_buffer_free(edge_conn->pending_optimistic_data);
}
if (edge_conn->sending_optimistic_data) {
generic_buffer_free(edge_conn->sending_optimistic_data);
}
rend_data_free(edge_conn->rend_data); rend_data_free(edge_conn->rend_data);
} }
if (conn->type == CONN_TYPE_CONTROL) { if (conn->type == CONN_TYPE_CONTROL) {

View File

@ -738,6 +738,12 @@ connection_ap_detach_retriable(edge_connection_t *conn, origin_circuit_t *circ,
{ {
control_event_stream_status(conn, STREAM_EVENT_FAILED_RETRIABLE, reason); control_event_stream_status(conn, STREAM_EVENT_FAILED_RETRIABLE, reason);
conn->_base.timestamp_lastread = time(NULL); conn->_base.timestamp_lastread = time(NULL);
if (conn->pending_optimistic_data) {
generic_buffer_set_to_copy(&conn->sending_optimistic_data,
conn->pending_optimistic_data);
}
if (!get_options()->LeaveStreamsUnattached || conn->use_begindir) { if (!get_options()->LeaveStreamsUnattached || conn->use_begindir) {
/* If we're attaching streams ourself, or if this connection is /* If we're attaching streams ourself, or if this connection is
* a tunneled directory connection, then just attach it. */ * a tunneled directory connection, then just attach it. */
@ -2429,7 +2435,8 @@ connection_ap_handshake_send_begin(edge_connection_t *ap_conn)
control_event_stream_status(ap_conn, STREAM_EVENT_SENT_CONNECT, 0); control_event_stream_status(ap_conn, STREAM_EVENT_SENT_CONNECT, 0);
/* If there's queued-up data, send it now */ /* If there's queued-up data, send it now */
if (connection_get_inbuf_len(TO_CONN(ap_conn)) && if ((connection_get_inbuf_len(TO_CONN(ap_conn)) ||
ap_conn->sending_optimistic_data) &&
connection_ap_supports_optimistic_data(ap_conn)) { connection_ap_supports_optimistic_data(ap_conn)) {
log_info(LD_APP, "Sending up to %ld bytes of queued-up data", log_info(LD_APP, "Sending up to %ld bytes of queued-up data",
connection_get_inbuf_len(TO_CONN(ap_conn))); connection_get_inbuf_len(TO_CONN(ap_conn)));

View File

@ -1237,11 +1237,20 @@ typedef struct edge_connection_t {
* NATd connection */ * NATd connection */
unsigned int is_transparent_ap:1; unsigned int is_transparent_ap:1;
/** Set if this connection's target exit node allows optimistic data. /** For AP connections only: Set if this connection's target exit node
* (That is, data sent on this stream before the exit has sent a * allows optimistic data. (That is, data sent on this stream before
* CONNECTED cell.)*/ * the exit has sent a CONNECTED cell.)*/
unsigned int exit_allows_optimistic_data : 1; unsigned int exit_allows_optimistic_data : 1;
/** For AP connections only: buffer for data that we have sent
* optimistically, which we might need to re-send if we have to
* retry this connection. */
generic_buffer_t *pending_optimistic_data;
/* For AP connections only: buffer for data that we previously sent
* optimistically which we are currently re-sending as we retry this
* connection. */
generic_buffer_t *sending_optimistic_data;
/** If this is a DNSPort connection, this field holds the pending DNS /** If this is a DNSPort connection, this field holds the pending DNS
* request that we're going to try to answer. */ * request that we're going to try to answer. */
struct evdns_server_request *dns_server_request; struct evdns_server_request *dns_server_request;

View File

@ -944,6 +944,12 @@ connection_edge_process_relay_cell_not_open(
break; break;
} }
} }
/* This is definitely a success, so forget about any pending data we
* had sent. */
if (conn->pending_optimistic_data) {
generic_buffer_free(conn->pending_optimistic_data);
conn->pending_optimistic_data = NULL;
}
/* handle anything that might have queued */ /* handle anything that might have queued */
if (connection_edge_package_raw_inbuf(conn, 1, NULL) < 0) { if (connection_edge_package_raw_inbuf(conn, 1, NULL) < 0) {
@ -1343,6 +1349,10 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
char payload[CELL_PAYLOAD_SIZE]; char payload[CELL_PAYLOAD_SIZE];
circuit_t *circ; circuit_t *circ;
unsigned domain = conn->cpath_layer ? LD_APP : LD_EXIT; unsigned domain = conn->cpath_layer ? LD_APP : LD_EXIT;
int sending_from_optimistic = 0;
const int sending_optimistically =
conn->_base.type == CONN_TYPE_AP &&
conn->_base.state != AP_CONN_STATE_OPEN;
tor_assert(conn); tor_assert(conn);
@ -1375,7 +1385,18 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
return 0; return 0;
} }
sending_from_optimistic = conn->sending_optimistic_data != NULL;
if (PREDICT_UNLIKELY(sending_from_optimistic)) {
amount_to_process = generic_buffer_len(conn->sending_optimistic_data);
if (PREDICT_UNLIKELY(!amount_to_process)) {
log_warn(LD_BUG, "sending_optimistic_data was non-NULL but empty");
amount_to_process = connection_get_inbuf_len(TO_CONN(conn)); amount_to_process = connection_get_inbuf_len(TO_CONN(conn));
sending_from_optimistic = 0;
}
} else {
amount_to_process = connection_get_inbuf_len(TO_CONN(conn));
}
if (!amount_to_process) if (!amount_to_process)
return 0; return 0;
@ -1391,11 +1412,30 @@ connection_edge_package_raw_inbuf(edge_connection_t *conn, int package_partial,
stats_n_data_bytes_packaged += length; stats_n_data_bytes_packaged += length;
stats_n_data_cells_packaged += 1; stats_n_data_cells_packaged += 1;
if (PREDICT_UNLIKELY(sending_from_optimistic)) {
/* XXX023 We could be more efficient here by sometimes packing
* previously-sent optimistic data in the same cell with data
* from the inbuf. */
generic_buffer_get(conn->sending_optimistic_data, payload, length);
if (!generic_buffer_len(conn->sending_optimistic_data)) {
generic_buffer_free(conn->sending_optimistic_data);
conn->sending_optimistic_data = NULL;
}
} else {
connection_fetch_from_buf(payload, length, TO_CONN(conn)); connection_fetch_from_buf(payload, length, TO_CONN(conn));
}
log_debug(domain,"(%d) Packaging %d bytes (%d waiting).", conn->_base.s, log_debug(domain,"(%d) Packaging %d bytes (%d waiting).", conn->_base.s,
(int)length, (int)connection_get_inbuf_len(TO_CONN(conn))); (int)length, (int)connection_get_inbuf_len(TO_CONN(conn)));
if (sending_optimistically && !sending_from_optimistic) {
/* This is new optimistic data; remember it in case we need to detach and
retry */
if (!conn->pending_optimistic_data)
conn->pending_optimistic_data = generic_buffer_new();
generic_buffer_add(conn->pending_optimistic_data, payload, length);
}
if (connection_edge_send_command(conn, RELAY_COMMAND_DATA, if (connection_edge_send_command(conn, RELAY_COMMAND_DATA,
payload, length) < 0 ) payload, length) < 0 )
/* circuit got marked for close, don't continue, don't need to mark conn */ /* circuit got marked for close, don't continue, don't need to mark conn */