1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/src/io/tx_link.c	Sat Feb 20 21:18:28 2010 -0800
     1.3 @@ -0,0 +1,279 @@
     1.4 +/*
     1.5 + * $Id: tx_link.c,v 1.10 2004/05/02 08:55:00 hipnod Exp $
     1.6 + *
     1.7 + * Copyright (C) 2003 giFT project (gift.sourceforge.net)
     1.8 + *
     1.9 + * This program is free software; you can redistribute it and/or modify it
    1.10 + * under the terms of the GNU General Public License as published by the
    1.11 + * Free Software Foundation; either version 2, or (at your option) any
    1.12 + * later version.
    1.13 + *
    1.14 + * This program is distributed in the hope that it will be useful, but
    1.15 + * WITHOUT ANY WARRANTY; without even the implied warranty of
    1.16 + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    1.17 + * General Public License for more details.
    1.18 + */
    1.19 +
    1.20 +#include "gt_gnutella.h"
    1.21 +
    1.22 +#include "io/tx_stack.h"
    1.23 +#include "io/tx_layer.h"
    1.24 +#include "io/io_buf.h"
    1.25 +
    1.26 +/*****************************************************************************/
    1.27 +
    1.28 +#define LINK_DEBUG 0
    1.29 +
    1.30 +/*****************************************************************************/
    1.31 +
    1.32 +struct tx_link
    1.33 +{
    1.34 +	input_id       id;
    1.35 +	struct io_buf *buf;
    1.36 +#if 0
    1.37 +	TCPC          *c;
    1.38 +#endif
    1.39 +};
    1.40 +
    1.41 +/*****************************************************************************/
    1.42 +
    1.43 +static void deactivate_queue    (struct tx_layer *tx);
    1.44 +static void activate_queue      (struct tx_layer *tx);
    1.45 +
    1.46 +/*****************************************************************************/
    1.47 +
    1.48 +static BOOL tx_link_init (struct tx_layer *tx)
    1.49 +{
    1.50 +	struct tx_link *tx_link;
    1.51 +
    1.52 +	if (!(tx_link = NEW (struct tx_link)))
    1.53 +		return FALSE;
    1.54 +
    1.55 +	tx_link->id  = 0;
    1.56 +	tx_link->buf = NULL;
    1.57 +
    1.58 +	/* store our layer-specific info in the toplevel layer */
    1.59 +	tx->udata = tx_link;
    1.60 +
    1.61 +	return TRUE;
    1.62 +}
    1.63 +
    1.64 +static void tx_link_destroy (struct tx_layer *tx)
    1.65 +{
    1.66 +	struct tx_link *tx_link = tx->udata;
    1.67 +
    1.68 +	input_remove (tx_link->id);
    1.69 +	tx_link->id = 0;
    1.70 +
    1.71 +	io_buf_free (tx_link->buf);
    1.72 +
    1.73 +	FREE (tx_link);
    1.74 +}
    1.75 +
    1.76 +/*****************************************************************************/
    1.77 +
    1.78 +static const char *tx_status_str (tx_status_t ret)
    1.79 +{
    1.80 +	switch (ret)
    1.81 +	{
    1.82 +	 case TX_EMPTY: return "TX_EMPTY";
    1.83 +	 case TX_FULL:  return "TX_FULL";
    1.84 +	 case TX_ERROR: return "TX_ERROR";
    1.85 +	 case TX_OK:    return "TX_OK";
    1.86 +	 case TX_PARTIAL: return "TX_PARTIAL";
    1.87 +	 default:       return "TX_UNKNOWN";
    1.88 +	}
    1.89 +}
    1.90 +
    1.91 +static tx_status_t request_more_data (struct tx_layer *tx)
    1.92 +{
    1.93 +	int ret;
    1.94 +
    1.95 +	/*
    1.96 +	 * Ask the layer above this one to send this layer data.
    1.97 +	 */
    1.98 +	if ((ret = gt_tx_layer_ready (tx)) == TX_ERROR)
    1.99 +		return TX_ERROR;
   1.100 +
   1.101 +	if (LINK_DEBUG)
   1.102 +		GT->DBGSOCK (GT, tx->stack->c, "ret=%s", tx_status_str (ret));
   1.103 +
   1.104 +	return ret;
   1.105 +}
   1.106 +
   1.107 +static tx_status_t tx_link_process (struct tx_layer *tx, struct io_buf *io_buf)
   1.108 +{
   1.109 +	uint8_t *ptr;
   1.110 +	size_t   len;
   1.111 +	int      n;
   1.112 +
   1.113 +	ptr = io_buf_read_ptr   (io_buf);
   1.114 +	len = io_buf_read_avail (io_buf);
   1.115 +
   1.116 +	/*
   1.117 +	 * gt_tx_stack_send() calls tcp_send() to send the data on the
   1.118 +	 * connection. This is done because no interface for passing parameters
   1.119 +	 * like a TCPC is exposed anywhere to users of GtTxStack.
   1.120 +	 */
   1.121 +	if ((n = gt_tx_stack_send (tx->stack, ptr, len)) <= 0)
   1.122 +		return TX_ERROR;
   1.123 +
   1.124 +	/*
   1.125 +	 * Pop whatever bytes were written off the buffer.  This may be less than
   1.126 +	 * the the whole buffer in the case of a short write.  In that case we
   1.127 +	 * don't remove the buffer, but continue writing it later.
   1.128 +	 */
   1.129 +	io_buf_pop (io_buf, n);
   1.130 +
   1.131 +	return TX_OK;
   1.132 +}
   1.133 +
   1.134 +/*
   1.135 + * The packet-sending input callback.
   1.136 + */
   1.137 +static void tx_link_send_data (int fd, input_id id, struct tx_layer *tx)
   1.138 +{
   1.139 +	struct tx_link *tx_link = tx->udata;
   1.140 +	struct io_buf  *io_buf;
   1.141 +	size_t          len;
   1.142 +
   1.143 +	/*
   1.144 +	 * If there's no data to write, request more from the upper layer.
   1.145 +	 */
   1.146 +	if (!(io_buf = tx_link->buf))
   1.147 +	{
   1.148 +		tx_status_t ret;
   1.149 +
   1.150 +		if ((ret = request_more_data (tx)) == TX_ERROR)
   1.151 +		{
   1.152 +			gt_tx_stack_abort (tx->stack);
   1.153 +			return;
   1.154 +		}
   1.155 +
   1.156 +		/* deactivate the queue until more data arrives */
   1.157 +		if (ret == TX_EMPTY)
   1.158 +		{
   1.159 +			if (LINK_DEBUG)
   1.160 +				GT->DBGSOCK (GT, tx->stack->c, "empty, deactivating");
   1.161 +
   1.162 +			assert (tx_link->buf == NULL);
   1.163 +			deactivate_queue (tx);
   1.164 +			return;
   1.165 +		}
   1.166 +
   1.167 +		/* upper layer must have called our queue function */
   1.168 +		assert (tx_link->buf != NULL);
   1.169 +		io_buf = tx_link->buf;
   1.170 +
   1.171 +		/* fall through and send the buffer */
   1.172 +	}
   1.173 +
   1.174 +	len = io_buf_read_avail (io_buf);
   1.175 +
   1.176 +	/*
   1.177 +	 * It is safe to abort the tx-stack if we encountered an error because
   1.178 +	 * there are no other callers into it currently.
   1.179 +	 */
   1.180 +	if (tx_link_process (tx, io_buf) == TX_ERROR)
   1.181 +	{
   1.182 +		gt_tx_stack_abort (tx->stack);
   1.183 +		return;
   1.184 +	}
   1.185 +
   1.186 +	if (io_buf_read_avail (io_buf) > 0)
   1.187 +	{
   1.188 +		assert (io_buf_read_avail (io_buf) < len);
   1.189 +		return;
   1.190 +	}
   1.191 +
   1.192 +	/*
   1.193 +	 * The complete buffer was written.  This input callback will continue
   1.194 +	 * grabbing data from the upper layer until gt_tx_layer_ready() returns
   1.195 +	 * TX_EMPTY or TX_ERROR.
   1.196 +	 */
   1.197 +	io_buf_free (io_buf);
   1.198 +	tx_link->buf = NULL;
   1.199 +}
   1.200 +
   1.201 +static void activate_queue (struct tx_layer *tx)
   1.202 +{
   1.203 +	struct tx_link *tx_link = tx->udata;
   1.204 +
   1.205 +	/* skip if input already active */
   1.206 +	if (tx_link->id)
   1.207 +		return;
   1.208 +
   1.209 +	tx_link->id = input_add (tx->stack->c->fd, tx, INPUT_WRITE,
   1.210 +	                         (InputCallback)tx_link_send_data, 0);
   1.211 +}
   1.212 +
   1.213 +static void deactivate_queue (struct tx_layer *tx)
   1.214 +{
   1.215 +	struct tx_link *tx_link = tx->udata;
   1.216 +
   1.217 +	if (!tx_link->id)
   1.218 +		return;
   1.219 +
   1.220 +	input_remove (tx_link->id);
   1.221 +	tx_link->id = 0;
   1.222 +}
   1.223 +
   1.224 +/* begin or end consuming data in this layer */
   1.225 +static void tx_link_toggle (struct tx_layer *tx, BOOL stop)
   1.226 +{
   1.227 +	if (stop)
   1.228 +		deactivate_queue (tx);
   1.229 +	else
   1.230 +		activate_queue (tx);
   1.231 +}
   1.232 +
   1.233 +/*****************************************************************************/
   1.234 +
   1.235 +static tx_status_t tx_link_queue (struct tx_layer *tx, struct io_buf *io_buf)
   1.236 +{
   1.237 +	struct tx_link *tx_link = tx->udata;
   1.238 +
   1.239 +	if (tx_link->buf != NULL)
   1.240 +	{
   1.241 +		/* this layer is "saturated" with its single packet */
   1.242 +		return TX_FULL;
   1.243 +	}
   1.244 +
   1.245 +	tx_link->buf = io_buf;
   1.246 +	activate_queue (tx);
   1.247 +
   1.248 +	/* TODO: need to change this if we change to writing as much
   1.249 +	 * as possible synchronously instead of one message from the handler */
   1.250 +	return TX_OK;
   1.251 +}
   1.252 +
   1.253 +static tx_status_t tx_link_ready (struct tx_layer *tx)
   1.254 +{
   1.255 +	abort (); /* can't handle layers underneath us */
   1.256 +	return TX_ERROR;
   1.257 +}
   1.258 +
   1.259 +/*****************************************************************************/
   1.260 +
   1.261 +static void tx_link_enable (struct tx_layer *tx)
   1.262 +{
   1.263 +	activate_queue (tx);
   1.264 +}
   1.265 +
   1.266 +static void tx_link_disable (struct tx_layer *tx)
   1.267 +{
   1.268 +	deactivate_queue (tx);
   1.269 +}
   1.270 +
   1.271 +/*****************************************************************************/
   1.272 +
   1.273 +struct tx_layer_ops gt_tx_link_ops =
   1.274 +{
   1.275 +	tx_link_init,
   1.276 +	tx_link_destroy,
   1.277 +	tx_link_toggle,
   1.278 +	tx_link_queue,
   1.279 +	tx_link_ready,
   1.280 +	tx_link_enable,
   1.281 +	tx_link_disable,
   1.282 +};