Merge branch 'optimistic-client' (eaa1c053) · Commits · Gaba / Tor

src/or/buffers.c

+70 −0

Original line number	Diff line number	Diff line
		@@ -557,6 +557,39 @@ buf_free(buf_t *buf)
		tor_free(buf);
		}

		/** Return a new copy of <b>in_chunk</b> */
		static chunk_t *
		chunk_copy(const chunk_t *in_chunk)
		{
		chunk_t *newch = tor_memdup(in_chunk, CHUNK_ALLOC_SIZE(in_chunk->memlen));
		newch->next = NULL;
		if (in_chunk->data) {
		off_t offset = in_chunk->data - in_chunk->mem;
		newch->data = newch->mem + offset;
		}
		return newch;
		}

		/** Return a new copy of <b>buf</b> */
		buf_t *
		buf_copy(const buf_t *buf)
		{
		chunk_t *ch;
		buf_t *out = buf_new();
		out->default_chunk_size = buf->default_chunk_size;
		for (ch = buf->head; ch; ch = ch->next) {
		chunk_t *newch = chunk_copy(ch);
		if (out->tail) {
		out->tail->next = newch;
		out->tail = newch;
		} else {
		out->head = out->tail = newch;
		}
		}
		out->datalen = buf->datalen;
		return out;
		}

		/** Append a new chunk with enough capacity to hold <b>capacity</b> bytes to
		* the tail of <b>buf</b>. If <b>capped</b>, don't allocate a chunk bigger
		* than MAX_CHUNK_ALLOC. */
		@@ -2374,6 +2407,43 @@ write_to_evbuffer_zlib(struct evbuffer buf, tor_zlib_state_t state,
		}
		#endif

		/** Set <b>output</b> to contain a copy of the data in <b>input</b> */
		int
		generic_buffer_set_to_copy(generic_buffer_t **output,
		const generic_buffer_t *input)
		{
		#ifdef USE_BUFFEREVENTS
		struct evbuffer_ptr ptr;
		size_t remaining = evbuffer_get_length(input);
		if (*output) {
		evbuffer_drain(output, evbuffer_get_length(output));
		} else {
		if (!(*output = evbuffer_new()))
		return -1;
		}
		evbuffer_ptr_set((struct evbuffer*)input, &ptr, 0, EVBUFFER_PTR_SET);
		while (remaining) {
		struct evbuffer_iovec v[4];
		int n_used, i;
		n_used = evbuffer_peek((struct evbuffer*)input, -1, &ptr, v, 4);
		if (n_used < 0)
		return -1;
		for (i=0;i<n_used;++i) {
		evbuffer_add(*output, v[i].iov_base, v[i].iov_len);
		tor_assert(v[i].iov_len <= remaining);
		remaining -= v[i].iov_len;
		evbuffer_ptr_set((struct evbuffer*)input,
		&ptr, v[i].iov_len, EVBUFFER_PTR_ADD);
		}
		}
		#else
		if (*output)
		buf_free(*output);
		*output = buf_copy(input);
		#endif
		return 0;
		}

		/** Log an error and exit if <b>buf</b> is corrupted.
		*/
		void

src/or/buffers.h

+19 −0

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ buf_t *buf_new(void);
		buf_t *buf_new_with_capacity(size_t size);
		void buf_free(buf_t *buf);
		void buf_clear(buf_t *buf);
		buf_t buf_copy(const buf_t buf);
		void buf_shrink(buf_t *buf);
		void buf_shrink_freelists(int free_all);
		void buf_dump_freelist_sizes(int severity);
		@@ -67,6 +68,24 @@ int write_to_evbuffer_zlib(struct evbuffer buf, tor_zlib_state_t state,
		int done);
		#endif

		#ifdef USE_BUFFEREVENTS
		#define generic_buffer_new() evbuffer_new()
		#define generic_buffer_len(b) evbuffer_get_length((b))
		#define generic_buffer_add(b,dat,len) evbuffer_add((b),(dat),(len))
		#define generic_buffer_get(b,buf,buflen) evbuffer_remove((b),(buf),(buflen))
		#define generic_buffer_clear(b) evbuffer_drain((b), evbuffer_get_length((b)))
		#define generic_buffer_free(b) evbuffer_free((b))
		#else
		#define generic_buffer_new() buf_new()
		#define generic_buffer_len(b) buf_datalen((b))
		#define generic_buffer_add(b,dat,len) write_to_buf((dat),(len),(b))
		#define generic_buffer_get(b,buf,buflen) fetch_from_buf((buf),(buflen),(b))
		#define generic_buffer_clear(b) buf_clear((b))
		#define generic_buffer_free(b) buf_free((b))
		#endif
		int generic_buffer_set_to_copy(generic_buffer_t **output,
		const generic_buffer_t *input);

		void assert_buf_ok(buf_t *buf);

		#ifdef BUFFERS_PRIVATE

src/or/circuituse.c

+21 −0

Original line number	Diff line number	Diff line
		@@ -738,6 +738,7 @@ circuit_detach_stream(circuit_t circ, edge_connection_t conn)
		tor_assert(conn);

		conn->cpath_layer = NULL; /* make sure we don't keep a stale pointer */
		conn->exit_allows_optimistic_data = 0;
		conn->on_circuit = NULL;

		if (CIRCUIT_IS_ORIGIN(circ)) {
		@@ -1548,6 +1549,8 @@ static void
		link_apconn_to_circ(edge_connection_t apconn, origin_circuit_t circ,
		crypt_path_t *cpath)
		{
		const node_t *exitnode;

		/* add it into the linked list of streams on this circuit */
		log_debug(LD_APP\|LD_CIRC, "attaching new conn to circ. n_circ_id %d.",
		circ->_base.n_circ_id);
		@@ -1570,6 +1573,24 @@ link_apconn_to_circ(edge_connection_t apconn, origin_circuit_t circ,

		circ->isolation_any_streams_attached = 1;
		connection_edge_update_circuit_isolation(apconn, circ, 0);

		/* See if we can use optimistic data on this circuit */
		if (apconn->cpath_layer->extend_info &&
		(exitnode = node_get_by_id(
		apconn->cpath_layer->extend_info->identity_digest)) &&
		exitnode->rs) {
		/* Okay; we know what exit node this is. */
		if (circ->_base.purpose == CIRCUIT_PURPOSE_C_GENERAL &&
		exitnode->rs->version_supports_optimistic_data)
		apconn->exit_allows_optimistic_data = 1;
		else
		apconn->exit_allows_optimistic_data = 0;
		log_info(LD_APP, "Looks like completed circuit to %s %s allow "
		"optimistic data for connection to %s",
		safe_str_client(node_describe(exitnode)),
		apconn->exit_allows_optimistic_data ? "does" : "doesn't",
		safe_str_client(apconn->socks_request->address));
		}
		}

		/** Return true iff <b>address</b> is matched by one of the entries in

src/or/connection.c

+6 −0

Original line number	Diff line number	Diff line
		@@ -470,6 +470,12 @@ _connection_free(connection_t *conn)
		tor_free(edge_conn->original_dest_address);
		if (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);
		}
		if (conn->type == CONN_TYPE_CONTROL) {

src/or/connection_edge.c

+51 −1

Original line number	Diff line number	Diff line
		@@ -57,6 +57,7 @@ static int connection_exit_connect_dir(edge_connection_t *exitconn);
		static int address_is_in_virtual_range(const char *addr);
		static int consider_plaintext_ports(edge_connection_t *conn, uint16_t port);
		static void clear_trackexithost_mappings(const char *exitname);
		static int connection_ap_supports_optimistic_data(const edge_connection_t *);

		/** An AP stream has failed/finished. If it hasn't already sent back
		* a socks reply, send one now (based on endreason). Also set
		@@ -154,10 +155,25 @@ connection_edge_process_inbuf(edge_connection_t *conn, int package_partial)
		return -1;
		}
		return 0;
		case AP_CONN_STATE_CONNECT_WAIT:
		if (connection_ap_supports_optimistic_data(conn)) {
		log_info(LD_EDGE,
		"data from edge while in '%s' state. Sending it anyway. "
		"package_partial=%d, buflen=%ld",
		conn_state_to_string(conn->_base.type, conn->_base.state),
		package_partial, connection_get_inbuf_len(TO_CONN(conn)));
		if (connection_edge_package_raw_inbuf(conn, package_partial, NULL)<0) {
		/* (We already sent an end cell if possible) */
		connection_mark_for_close(TO_CONN(conn));
		return -1;
		}
		return 0;
		}
		/* Fall through if the connection is on a circuit without optimistic
		* data support. */
		case EXIT_CONN_STATE_CONNECTING:
		case AP_CONN_STATE_RENDDESC_WAIT:
		case AP_CONN_STATE_CIRCUIT_WAIT:
		case AP_CONN_STATE_CONNECT_WAIT:
		case AP_CONN_STATE_RESOLVE_WAIT:
		case AP_CONN_STATE_CONTROLLER_WAIT:
		log_info(LD_EDGE,
		@@ -722,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);
		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 we're attaching streams ourself, or if this connection is
		* a tunneled directory connection, then just attach it. */
		@@ -2337,6 +2359,22 @@ get_unique_stream_id_by_circ(origin_circuit_t *circ)
		return test_stream_id;
		}

		/** Return true iff <b>conn</b> is linked to a circuit and configured to use
		* an exit that supports optimistic data. */
		static int
		connection_ap_supports_optimistic_data(const edge_connection_t *conn)
		{
		tor_assert(conn->_base.type == CONN_TYPE_AP);
		/* We can only send optimistic data if we're connected to an open
		general circuit. */
		if (conn->on_circuit == NULL \|\|
		conn->on_circuit->state != CIRCUIT_STATE_OPEN \|\|
		conn->on_circuit->purpose != CIRCUIT_PURPOSE_C_GENERAL)
		return 0;

		return conn->exit_allows_optimistic_data;
		}

		/** Write a relay begin cell, using destaddr and destport from ap_conn's
		* socks_request field, and send it down circ.
		*
		@@ -2398,6 +2436,18 @@ connection_ap_handshake_send_begin(edge_connection_t *ap_conn)
		log_info(LD_APP,"Address/port sent, ap socket %d, n_circ_id %d",
		ap_conn->_base.s, circ->_base.n_circ_id);
		control_event_stream_status(ap_conn, STREAM_EVENT_SENT_CONNECT, 0);

		/* If there's queued-up data, send it now */
		if ((connection_get_inbuf_len(TO_CONN(ap_conn)) \|\|
		ap_conn->sending_optimistic_data) &&
		connection_ap_supports_optimistic_data(ap_conn)) {
		log_info(LD_APP, "Sending up to %ld bytes of queued-up data",
		connection_get_inbuf_len(TO_CONN(ap_conn)));
		if (connection_edge_package_raw_inbuf(ap_conn, 1, NULL) < 0) {
		connection_mark_for_close(TO_CONN(ap_conn));
		}
		}

		return 0;
		}