[or-cvs] Add code for end-to-end zlib compression. Still needs flow...
Nick Mathewson
nickm at seul.org
Mon Mar 17 02:42:47 UTC 2003
Update of /home/or/cvsroot/src/or
In directory moria.mit.edu:/tmp/cvs-serv2671/src/or
Modified Files:
buffers.c connection.c connection_ap.c connection_exit.c or.h
Log Message:
Add code for end-to-end zlib compression. Still needs flow-control
Index: buffers.c
===================================================================
RCS file: /home/or/cvsroot/src/or/buffers.c,v
retrieving revision 1.12
retrieving revision 1.13
diff -u -d -r1.12 -r1.13
--- buffers.c 10 Mar 2003 22:40:02 -0000 1.12
+++ buffers.c 17 Mar 2003 02:42:45 -0000 1.13
@@ -136,6 +136,74 @@
}
+#ifdef USE_ZLIB
+int compress_from_buf(char *string, int string_len,
+ char **buf_in, int *buflen_in, int *buf_datalen_in,
+ z_stream *zstream, int flush) {
+ int err;
+
+ if (!*buf_datalen_in)
+ return 0;
+
+ zstream->next_in = *buf_in;
+ zstream->avail_in = *buf_datalen_in;
+ zstream->next_out = string;
+ zstream->avail_out = string_len;
+
+ err = deflate(zstream, flush);
+
+ switch (err)
+ {
+ case Z_OK:
+ case Z_STREAM_END:
+ memmove(*buf_in, zstream->next_in, zstream->avail_in);
+ *buf_datalen_in = zstream->avail_in;
+ return string_len - zstream->avail_out;
+ case Z_STREAM_ERROR:
+ case Z_BUF_ERROR:
+ log(LOG_ERR, "Error processing compression: %s", zstream->msg);
+ return -1;
+ default:
+ log(LOG_ERR, "Unknown return value from deflate: %d", err);
+ return -1;
+ }
+}
+
+int decompress_buf_to_buf(char **buf_in, int *buflen_in, int *buf_datalen_in,
+ char **buf_out, int *buflen_out, int *buf_datalen_out,
+ z_stream *zstream, int flush)
+{
+ int err;
+
+ zstream->next_in = *buf_in;
+ zstream->avail_in = *buf_datalen_in;
+ zstream->next_out = *buf_out + *buf_datalen_out;
+ zstream->avail_out = *buflen_out - *buf_datalen_out;
+
+ if (!zstream->avail_in && !zstream->avail_out)
+ return 0;
+
+ err = inflate(zstream, flush);
+
+ switch (err)
+ {
+ case Z_OK:
+ case Z_STREAM_END:
+ memmove(*buf_in, zstream->next_in, zstream->avail_in);
+ *buf_datalen_in = zstream->avail_in;
+ *buf_datalen_out = *buflen_out - zstream->avail_out;
+ return 1;
+ case Z_STREAM_ERROR:
+ case Z_BUF_ERROR:
+ log(LOG_ERR, "Error processing compression: %s", zstream->msg);
+ return 1;
+ default:
+ log(LOG_ERR, "Unknown return value from deflate: %d", err);
+ return -1;
+ }
+}
+#endif
+
int fetch_from_buf(char *string, int string_len,
char **buf, int *buflen, int *buf_datalen) {
Index: connection.c
===================================================================
RCS file: /home/or/cvsroot/src/or/connection.c,v
retrieving revision 1.41
retrieving revision 1.42
diff -u -d -r1.41 -r1.42
--- connection.c 12 Mar 2003 12:06:54 -0000 1.41
+++ connection.c 17 Mar 2003 02:42:45 -0000 1.42
@@ -126,6 +126,29 @@
if(type == CONN_TYPE_OR) {
directory_set_dirty();
}
+#ifdef USE_ZLIB
+ if (type == CONN_TYPE_AP || type == CONN_TYPE_EXIT) {
+ if (buf_new(&conn->z_outbuf, &conn->z_outbuflen, &conn->z_outbuf_datalen) < 0)
+ return NULL;
+ if (! (conn->compression = malloc(sizeof(z_stream))))
+ return NULL;
+ if (! (conn->decompression = malloc(sizeof(z_stream))))
+ return NULL;
+ memset(conn->compression, 0, sizeof(z_stream));
+ memset(conn->decompression, 0, sizeof(z_stream));
+ if (deflateInit(conn->compression, Z_DEFAULT_COMPRESSION) != Z_OK) {
+ log(LOG_ERR, "Error initializing zlib: %s", conn->compression->msg);
+ return NULL;
+ }
+ if (inflateInit(conn->decompression) != Z_OK) {
+ log(LOG_ERR, "Error initializing zlib: %s", conn->decompression->msg);
+ return NULL;
+ }
+ } else {
+ conn->compression = conn->decompression = NULL;
+ }
+ conn->done_sending = conn->done_receiving = 0
+#endif
return conn;
}
@@ -156,6 +179,19 @@
if(conn->type == CONN_TYPE_OR) {
directory_set_dirty();
}
+#ifdef USE_ZLIB
+ if (conn->compression) {
+ if (inflateEnd(conn->decompression) != Z_OK)
+ log(LOG_ERR,"connection_free(): while closing zlib: %s",
+ conn->decompression->msg);
+ if (deflateEnd(conn->compression) != Z_OK)
+ log(LOG_ERR,"connection_free(): while closing zlib: %s",
+ conn->compression->msg);
+ free(conn->compression);
+ free(conn->decompression);
+ buf_free(conn->z_outbuf);
+ }
+#endif
free(conn);
}
@@ -337,6 +373,49 @@
return fetch_from_buf(string, len, &conn->inbuf, &conn->inbuflen, &conn->inbuf_datalen);
}
+#ifdef USE_ZLIB
+int connection_compress_from_buf(char *string, int len, connection_t *conn,
+ int flush) {
+ return compress_from_buf(string, len,
+ &conn->inbuf, &conn->inbuflen, &conn->inbuf_datalen,
+ conn->compression, flush);
+}
+
+int connection_decompress_to_buf(char *string, int len, connection_t *conn,
+ int flush) {
+ /* This is not sane with respect to flow control; we want to spool out to
+ * z_outbuf, but only decompress and write as needed.
+ */
+ int n;
+ struct timeval now;
+
+ if (write_to_buf(string, len,
+ &conn->z_outbuf, &conn->z_outbuflen, &conn->z_outbuf_datalen) < 0)
+ return -1;
+
+ n = decompress_buf_to_buf(
+ &conn->z_outbuf, &conn->z_outbuflen, &conn->z_outbuf_datalen,
+ &conn->outbuf, &conn->outbuflen, &conn->outbuf_datalen,
+ conn->decompression, flush);
+
+ if (n < 0)
+ return -1;
+
+ if(gettimeofday(&now,NULL) < 0)
+ return -1;
+
+ if(!n)
+ return 0;
+
+ if(conn->marked_for_close)
+ return 0;
+
+ conn->timestamp_lastwritten = now.tv_sec;
+
+ return n;
+}
+#endif
+
int connection_find_on_inbuf(char *string, int len, connection_t *conn) {
return find_on_inbuf(string, len, conn->inbuf, conn->inbuf_datalen);
}
@@ -607,7 +686,7 @@
}
int connection_package_raw_inbuf(connection_t *conn) {
- int amount_to_process;
+ int amount_to_process, len;
cell_t cell;
circuit_t *circ;
@@ -618,13 +697,27 @@
repeat_connection_package_raw_inbuf:
amount_to_process = conn->inbuf_datalen;
-
+
if(!amount_to_process)
return 0;
/* Initialize the cell with 0's */
memset(&cell, 0, sizeof(cell_t));
+#ifdef USE_ZLIB
+ /* This compression logic is not necessarily optimal:
+ * 1) Maybe we should try to read as much as we can onto the inbuf before
+ * compressing.
+ * 2)
+ */
+ len = connection_compress_from_buf(cell.payload + TOPIC_HEADER_SIZE,
+ CELL_PAYLOAD_SIZE - TOPIC_HEADER_SIZE,
+ conn, Z_SYNC_FLUSH);
+ if (len < 0)
+ return -1;
+
+ cell.length = len;
+#else
if(amount_to_process > CELL_PAYLOAD_SIZE - TOPIC_HEADER_SIZE) {
cell.length = CELL_PAYLOAD_SIZE - TOPIC_HEADER_SIZE;
} else {
@@ -633,6 +726,7 @@
if(connection_fetch_from_buf(cell.payload+TOPIC_HEADER_SIZE, cell.length, conn) < 0)
return -1;
+#endif
circ = circuit_get_by_conn(conn);
if(!circ) {
@@ -677,7 +771,7 @@
}
log(LOG_DEBUG,"connection_package_raw_inbuf(): receive_topicwindow at AP is %d",conn->p_receive_topicwindow);
}
- if(amount_to_process > CELL_PAYLOAD_SIZE - TOPIC_HEADER_SIZE) {
+ if (conn->inbuf_datalen) {
log(LOG_DEBUG,"connection_package_raw_inbuf(): recursing.");
goto repeat_connection_package_raw_inbuf;
}
Index: connection_ap.c
===================================================================
RCS file: /home/or/cvsroot/src/or/connection_ap.c,v
retrieving revision 1.26
retrieving revision 1.27
diff -u -d -r1.26 -r1.27
--- connection_ap.c 11 Mar 2003 21:38:38 -0000 1.26
+++ connection_ap.c 17 Mar 2003 02:42:45 -0000 1.27
@@ -418,11 +418,21 @@
}
log(LOG_DEBUG,"connection_ap_process_data_cell(): willing to receive %d more cells from circ",conn->n_receive_topicwindow);
+#ifdef USE_ZLIB
+ if(connection_decompress_to_buf(cell->payload + TOPIC_HEADER_SIZE,
+ cell->length - TOPIC_HEADER_SIZE,
+ conn, Z_SYNC_FLUSH) < 0) {
+ log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
+ conn->marked_for_close = 1;
+ return 0;
+ }
+#else
if(connection_write_to_buf(cell->payload + TOPIC_HEADER_SIZE,
cell->length - TOPIC_HEADER_SIZE, conn) < 0) {
conn->marked_for_close = 1;
return 0;
}
+#endif
if(connection_consider_sending_sendme(conn, EDGE_AP) < 0)
conn->marked_for_close = 1;
return 0;
@@ -440,6 +450,12 @@
}
for(prevconn = circ->p_conn; prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
prevconn->next_topic = conn->next_topic;
+#endif
+#if 0
+ conn->done_sending = 1;
+ shutdown(conn->s, 1); /* XXX check return; refactor NM */
+ if (conn->done_receiving)
+ conn->marked_for_close = 1;
#endif
conn->marked_for_close = 1;
break;
Index: connection_exit.c
===================================================================
RCS file: /home/or/cvsroot/src/or/connection_exit.c,v
retrieving revision 1.21
retrieving revision 1.22
diff -u -d -r1.21 -r1.22
--- connection_exit.c 18 Feb 2003 01:35:55 -0000 1.21
+++ connection_exit.c 17 Mar 2003 02:42:45 -0000 1.22
@@ -217,12 +217,22 @@
log(LOG_DEBUG,"connection_exit_process_data_cell(): data received while resolving/connecting. Queueing.");
}
log(LOG_DEBUG,"connection_exit_process_data_cell(): put %d bytes on outbuf.",cell->length - TOPIC_HEADER_SIZE);
+#ifdef USE_ZLIB
+ if(connection_decompress_to_buf(cell->payload + TOPIC_HEADER_SIZE,
+ cell->length - TOPIC_HEADER_SIZE,
+ conn, Z_SYNC_FLUSH) < 0) {
+ log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
+ conn->marked_for_close = 1;
+ return 0;
+ }
+#else
if(connection_write_to_buf(cell->payload + TOPIC_HEADER_SIZE,
cell->length - TOPIC_HEADER_SIZE, conn) < 0) {
log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
conn->marked_for_close = 1;
return 0;
}
+#endif
if(connection_consider_sending_sendme(conn, EDGE_EXIT) < 0)
conn->marked_for_close = 1;
return 0;
@@ -241,6 +251,13 @@
for(prevconn = circ->n_conn; prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
prevconn->next_topic = conn->next_topic;
#endif
+#if 0
+ conn->done_sending = 1;
+ shutdown(conn->s, 1); /* XXX check return; refactor NM */
+ if (conn->done_receiving)
+ conn->marked_for_close = 1;
+#endif
+
conn->marked_for_close = 1;
break;
case TOPIC_COMMAND_CONNECTED:
Index: or.h
===================================================================
RCS file: /home/or/cvsroot/src/or/or.h,v
retrieving revision 1.50
retrieving revision 1.51
diff -u -d -r1.50 -r1.51
--- or.h 11 Mar 2003 01:51:41 -0000 1.50
+++ or.h 17 Mar 2003 02:42:45 -0000 1.51
@@ -36,6 +36,9 @@
#include <errno.h>
#include <assert.h>
#include <time.h>
+#ifdef USE_ZLIB
+#include <zlib.h>
+#endif
#include "../common/crypto.h"
#include "../common/log.h"
@@ -171,6 +174,7 @@
#define CONFIG_TYPE_INT 2
#define CONFIG_TYPE_LONG 3
#define CONFIG_TYPE_DOUBLE 4
+#define CONFIG_TYPE_BOOL 5
#define CONFIG_LINE_MAXLEN 1024
@@ -254,24 +258,39 @@
uint16_t port;
/* used by exit and ap: */
-
uint16_t topic_id;
struct connection_t *next_topic;
int n_receive_topicwindow;
int p_receive_topicwindow;
+ int done_sending;
+ int done_receiving;
+#ifdef USE_ZLIB
+ char *z_outbuf;
+ int z_outbuflen;
+ int z_outbuf_datalen;
+ z_stream *compression;
+ z_stream *decompression;
+#endif
+
+/* Used by ap: */
char socks_version;
char read_username;
+/* Used by exit and ap: */
char *dest_addr;
uint16_t dest_port; /* host order */
+/* Used by ap: */
char dest_tmp[512];
int dest_tmplen;
+/* Used by everyone */
char *address; /* strdup into this, because free_connection frees it */
+/* Used for cell connections */
crypto_pk_env_t *pkey; /* public RSA key for the other side */
+/* Used while negotiating OR/OR connections */
char nonce[8];
};
@@ -383,6 +402,7 @@
char *RouterFile;
char *PrivateKeyFile;
double CoinWeight;
+ int Daemon;
int ORPort;
int OPPort;
int APPort;
@@ -421,12 +441,29 @@
* return total number of bytes on the buf
*/
+
int fetch_from_buf(char *string, int string_len,
char **buf, int *buflen, int *buf_datalen);
/* if there is string_len bytes in buf, write them onto string,
* then memmove buf back (that is, remove them from buf)
*/
+#ifdef USE_ZLIB
+int compress_from_buf(char *string, int string_len,
+ char **buf_in, int *buflen_in, int *buf_datalen_in,
+ z_stream *zstream, int flush);
+ /* read and compress as many characters as possible from buf, writing up to
+ * string_len of them onto string, then memmove buf back. Return number of
+ * characters written.
+ */
+
+int decompress_buf_to_buf(char **buf_in, int *buflen_in, int *buf_datalen_in,
+ char **buf_out, int *buflen_out, int *buf_datalen_out,
+ z_stream *zstream, int flush);
+ /* XXX document this NM
+ */
+#endif
+
int find_on_inbuf(char *string, int string_len,
char *buf, int buf_datalen);
/* find first instance of needle 'string' on haystack 'buf'. return how
@@ -528,6 +565,13 @@
int connection_read_to_buf(connection_t *conn);
int connection_fetch_from_buf(char *string, int len, connection_t *conn);
+
+#ifdef USE_ZLIB
+int connection_compress_from_buf(char *string, int len, connection_t *conn,
+ int flush);
+int connection_decompress_to_buf(char *string, int len, connection_t *conn,
+ int flush);
+#endif
int connection_outbuf_too_full(connection_t *conn);
int connection_find_on_inbuf(char *string, int len, connection_t *conn);
More information about the tor-commits
mailing list