implement buffering to chunks in E1->application direction

Existing applications (such as those written for DAHDI) expect to
be reading data in buffer/chunk sizes.  For example OsmoNITB: It
doesn't want to execute an expensive read/recv syscall to receive 11
bytes, if it needs at least 160 bytes.

Change-Id: I807671bc6f2acaef740ce215b8d8abcb5dce2640

src/ctl.c

@@ -121,6 +121,20 @@ e1_ts_stop(struct e1_ts *ts)
 		close(ts->fd);
 		ts->fd = -1;
 	}
+
+	talloc_free(ts->raw.rx_buf);
+	ts->raw.rx_buf = NULL;
+	ts->raw.rx_buf_size = 0;
+	ts->raw.rx_buf_used = 0;
+}
+
+static void
+_e1d_ts_raw_buf_realloc(struct e1_ts *ts, unsigned int size)
+{
+	ts->raw.rx_buf = talloc_realloc_size(ts->line, ts->raw.rx_buf, size);
+	OSMO_ASSERT(ts->raw.rx_buf);
+	ts->raw.rx_buf_size = size;
+	ts->raw.rx_buf_used = 0;
 }
 
 static int
@@ -137,6 +151,7 @@ _e1d_ts_start(struct e1_ts *ts, enum e1_ts_mode mode)
 		break;
 	case E1_TS_MODE_RAW:
 		sock_type = SOCK_STREAM;
+		_e1d_ts_raw_buf_realloc(ts, 160); /* TODO: make configurable */
 		break;
 	default:
 		return -EINVAL;

src/e1d.h

@@ -53,6 +53,13 @@ struct e1_ts {
 		int tx_len;
 	} hdlc;
 
+	/* RAW handling */
+	struct {
+		uint8_t *rx_buf;		/* actual buffer storage */
+		unsigned int rx_buf_size;	/* size of 'buf' in bytes */
+		unsigned int rx_buf_used;	/* number of bytes used so far */
+	} raw;
+
 	/* Remote end */
 	int fd;
 };

src/intf_line.c

@@ -200,7 +200,7 @@ _e1_tx_hdlcfs(struct e1_ts *ts, uint8_t *buf, int len)
 				}
 				LOGPTS(ts, DXFR, LOGL_DEBUG, "TX Message: %d [ %s]\n",
 					rv, osmo_hexdump(ts->hdlc.tx_buf, rv));
-				ts->hdlc.tx_len = rv; 
+				ts->hdlc.tx_len = rv;
 				ts->hdlc.tx_ofs = 0;
 			} else if (rv < 0 && errno != EAGAIN)
 				return rv;
@@ -338,6 +338,48 @@ e1_line_mux_out(struct e1_line *line, uint8_t *buf, int fts)
 	return tsz;
 }
 
+/* append data to the per-timeslot buffer; flush to socket every time buffer is full */
+static int
+_e1_rx_raw(struct e1_ts *ts, const uint8_t *buf, unsigned int len)
+{
+	unsigned int appended = 0;
+	int rv;
+
+	OSMO_ASSERT(ts->mode == E1_TS_MODE_RAW);
+
+	/* we don't keep a larger set of buffers but simply assume that whenever
+	 * we received one full chunk/buffer size, we are able to push the data
+	 * into the underlying unix domain socket.  Kernel socket buffering should
+	 * be far sufficient in terms of buffering capacity of voice data (which
+	 * is typically consumed reasonably low latency and hence buffer size) */
+
+	while (appended < len) {
+		unsigned int ts_buf_tailroom = ts->raw.rx_buf_size - ts->raw.rx_buf_used;
+		unsigned int chunk_len;
+
+		/* determine size of chunk we can write at this point */
+		chunk_len = len - appended;
+		if (chunk_len > ts_buf_tailroom)
+			chunk_len = ts_buf_tailroom;
+
+		/* actually copy the chunk */
+		memcpy(ts->raw.rx_buf + ts->raw.rx_buf_used, buf + appended, chunk_len);
+		ts->raw.rx_buf_used += chunk_len;
+		appended += chunk_len;
+
+		/* if ts_buf is full: flush + rewind */
+		if (ts->raw.rx_buf_used >= ts->raw.rx_buf_size) {
+			rv = write(ts->fd, ts->raw.rx_buf, ts->raw.rx_buf_size);
+			if (rv < 0)
+				return rv;
+			/* FIXME: count overflows */
+			ts->raw.rx_buf_used = 0;
+		}
+	}
+
+	return appended;
+}
+
 /* write data to a timeslot (hardware -> application direction) */
 static int
 _e1_ts_write(struct e1_ts *ts, const uint8_t *buf, size_t len)
@@ -346,7 +388,7 @@ _e1_ts_write(struct e1_ts *ts, const uint8_t *buf, size_t len)
 
 	switch (ts->mode) {
 	case E1_TS_MODE_RAW:
-		rv = write(ts->fd, buf, len);
+		rv = _e1_rx_raw(ts, buf, len);
 		break;
 	case E1_TS_MODE_HDLCFCS:
 		rv = _e1_rx_hdlcfs(ts, buf, len);