qse/cmd/http/ursd.c

#include <qse/cmn/main.h>
#include <qse/cmn/str.h>
#include <qse/cmn/mem.h>
#include <qse/cmn/mbwc.h>
#include <qse/cmn/time.h>
#include <qse/cmn/path.h>
#include <qse/cmn/opt.h>
#include <qse/cmn/htb.h>
#include <qse/cmn/fmt.h>
#include <qse/cmn/hton.h>

#include <qse/si/sck.h>
#include <qse/si/nwad.h>
#include <qse/si/sio.h>
#include <qse/si/pio.h>
#include <qse/si/mux.h>

#include <signal.h>
#include <locale.h>

#if defined(_WIN32)
#	include <winsock2.h>
#	include <windows.h>
#	include <tchar.h>
#	include <process.h>
#elif defined(__OS2__)
#	define INCL_DOSPROCESS
#	define INCL_DOSEXCEPTIONS
#	define INCL_ERRORS
#	include <os2.h>
#	if defined(TCPV40HDRS)
#		define  BSD_SELECT
#	endif
#	include <types.h>
#	include <sys/socket.h>
#	include <netinet/in.h>
#	include <sys/ioctl.h>
#	include <nerrno.h>
#	if defined(TCPV40HDRS)
#		define  USE_SELECT
#		include <sys/select.h>
#	else
#		include <unistd.h>
#	endif
#elif defined(__DOS__)
#	include <dos.h>
#else
#	include <unistd.h>
#	include <errno.h>
#	include <fcntl.h>
#	include <sys/socket.h>
#	include <netinet/in.h>
#	if defined(HAVE_NETINET_SCTP_H)
#		include <netinet/sctp.h>
#	endif
#endif

#if defined(HAVE_SYS_PRCTL_H)
#	include <sys/prctl.h>
#endif

#if defined(HAVE_SYS_TIME_H)
#	include <sys/time.h>
#endif

#if defined(HAVE_SYS_RESOURCE_H)
#	include <sys/resource.h>
#endif

typedef struct urs_hdr_t urs_hdr_t;
typedef struct urs_pkt_t urs_pkt_t;

#define URS_RCODE_OK 0
#define URS_RCODE_ERROR 1

#include <qse/pack1.h>
struct urs_hdr_t
{
	qse_uint16_t seq; /* in network-byte order */
	qse_uint16_t rcode; /* response code */
	qse_uint32_t urlsum;/* checksum of url in the request */
	qse_uint16_t pktlen; /* url length in network-byte order */
};

struct urs_pkt_t
{
	struct urs_hdr_t hdr;
	qse_mchar_t url[1];
};
#include <qse/unpack.h>

#define MAX_PACKET_SIZE 65535
#define XREQ_BLOCK_SIZE 2048
#define XREQ_MAX_BLOCKS 32 

typedef struct xreq_t xreq_t;
struct xreq_t
{
	qse_ntime_t timestamp;

	qse_skad_t from;
	qse_sck_len_t fromlen;
	urs_pkt_t* pkt;

	xreq_t* next;
};

typedef struct rewriter_t rewriter_t;
struct rewriter_t
{
	int index; /* index in ursd->rewriters */
	qse_pio_t* pio;

	/* ------------------ */

	unsigned int free: 1;
	unsigned int busy: 1;
	unsigned int faulty: 1;
	unsigned int pio_in_in_mux: 1;

	struct
	{
		qse_skad_t from;
		qse_sck_len_t fromlen;
		qse_uint16_t urllen;
		qse_uint16_t urlpos;
		qse_uint8_t buf[MAX_PACKET_SIZE];
		qse_uint32_t outlen; /* length of output read from the rewriter */
	} req;

	rewriter_t* prev;
	rewriter_t* next;
};

typedef struct ursd_t ursd_t;
struct ursd_t
{
	qse_mmgr_t* mmgr;
	qse_char_t* cmdline;

	qse_size_t total_rewriter_count;
	qse_size_t free_rewriter_count;

	rewriter_t* rewriters;
	rewriter_t* free_rewriter;
	rewriter_t* busy_rewriter;

	qse_sck_hnd_t sck;
	qse_mux_t* mux;

	
	struct
	{
		qse_uint8_t buf[MAX_PACKET_SIZE]; /* temporary buffer */

		qse_size_t count; /* number of packets in the request queue */
		xreq_t* head; /* head of the request queue */
		xreq_t* tail; /* tail of the request queue */

		xreq_t* free[XREQ_MAX_BLOCKS]; /* xreq avaialble chains per block size */
	} xreq; /* request queue */

};


#define TYPE_SOCKET  0
#define TYPE_PIO_OUT 1
#define TYPE_PIO_IN  2

#define MAKE_MUX_DATA(type,index)   ((qse_uintptr_t)type | ((qse_uintptr_t)index << 4))
#define GET_TYPE_FROM_MUX_DATA(md)  ((md) & 0xF)
#define GET_INDEX_FROM_MUX_DATA(md) ((md) >> 4)

struct mux_xtn_t
{
	ursd_t* ursd;
};
typedef struct mux_xtn_t mux_xtn_t;

static qse_sck_hnd_t open_server_socket (int proto, const qse_nwad_t* bindnwad)
{
	qse_sck_hnd_t s = QSE_INVALID_SCKHND;
	qse_skad_t skad;
	qse_sck_len_t skad_len;
	int family, type, flag;

	skad_len = qse_nwadtoskad (bindnwad, &skad);
	family = qse_skadfamily(&skad);

#if defined(IPPROTO_SCTP)
	type = (proto == IPPROTO_SCTP)? SOCK_SEQPACKET: SOCK_DGRAM;
#else
	type = SOCK_DGRAM;
#endif

	if (bindnwad->type == QSE_NWAD_LOCAL) 
	{
		proto = 0;
		/* TODO: delete sun_path */
	}

	s = socket (family, type, proto);
	if (!qse_is_sck_valid(s))
	{
		qse_fprintf (QSE_STDERR, QSE_T("cannot create a socket\n"));
		goto oops;
	}

/* TODO: increase the socket buffer size, especially the output buffer size */

	#if defined(FD_CLOEXEC)
	flag = fcntl (s, F_GETFD);
	if (flag >= 0) fcntl (s, F_SETFD, flag | FD_CLOEXEC);
	#endif

	#if defined(SO_REUSEADDR)
	flag = 1;
	setsockopt (s, SOL_SOCKET, SO_REUSEADDR, (void*)&flag, QSE_SIZEOF(flag));
	#endif

	if (bind (s, (struct sockaddr*)&skad, skad_len) <= -1) 
	{
	#if defined(IPV6_V6ONLY) && defined(EADDRINUSE)
		if (errno == EADDRINUSE && family == AF_INET6)
		{
			int on = 1;
			setsockopt (s, IPPROTO_IPV6, IPV6_V6ONLY, &on, sizeof(on));
			if (bind (s, (struct sockaddr*)&skad, skad_len) == 0) goto bind_ok;
		}
		
	#endif
		qse_fprintf (QSE_STDERR, QSE_T("cannot bind a socket\n"));
		goto oops;
		
	}
	else
	{
		qse_char_t buf[512];
		qse_nwadtostr (bindnwad, buf, QSE_COUNTOF(buf), QSE_NWADTOSTR_ALL);
		qse_printf (QSE_T("binding address: %s\n"), buf);
	}

bind_ok:
	#if defined(IPPROTO_SCTP)
	if (proto == IPPROTO_SCTP)
	{
		#if defined(SOL_SCTP)
		struct sctp_initmsg im;
		struct sctp_paddrparams hb;

		qse_memset (&im, 0, QSE_SIZEOF(im));
		im.sinit_num_ostreams = 1;
		im.sinit_max_instreams = 1;
		im.sinit_max_attempts = 1;

		if (setsockopt (s, SOL_SCTP, SCTP_INITMSG, &im, QSE_SIZEOF(im)) <= -1) 
		{
			qse_fprintf (QSE_STDERR, QSE_T("cannot set sctp initmsg option\n"));
			goto oops;
		}

		qse_memset (&hb, 0, QSE_SIZEOF(hb));
		hb.spp_flags = SPP_HB_ENABLE;
		hb.spp_hbinterval = 5000;
		hb.spp_pathmaxrxt = 1;

		if (setsockopt(s, SOL_SCTP, SCTP_PEER_ADDR_PARAMS, &hb, QSE_SIZEOF(hb)) <= -1) goto oops;
		#endif

		if (listen(s, 99) <= -1)
		{
			qse_fprintf (QSE_STDERR, QSE_T("cannot set listen on sctp socket\n"));
			goto oops;
		}
	}
	#endif

	return s;

oops:
	if (qse_is_sck_valid(s)) qse_close_sck (s);
	return QSE_INVALID_SCKHND;
}

static int insert_to_mux (qse_mux_t* mux, qse_mux_hnd_t handle, int type, int index)
{
	qse_mux_evt_t evt;

	qse_memset (&evt, 0, QSE_SIZEOF(evt));
	evt.hnd = handle;
	evt.mask = (type == TYPE_PIO_IN? QSE_MUX_OUT: QSE_MUX_IN);
	evt.data = MAKE_MUX_DATA(type, index);
	return qse_mux_insert (mux, &evt);
}

static int delete_from_mux (qse_mux_t* mux, qse_mux_hnd_t handle, int type, int index)
{
	qse_mux_evt_t evt;

	qse_memset (&evt, 0, QSE_SIZEOF(evt));
	evt.hnd = handle;
	evt.mask = (type == TYPE_PIO_IN? QSE_MUX_OUT: QSE_MUX_IN);
	evt.data = MAKE_MUX_DATA(type, index);
	return qse_mux_delete (mux, &evt);
}

static void chain_rewriter_to_free_list (ursd_t* ursd, rewriter_t* rewriter)
{
	/* attach a rewriter to the head of the list */
	rewriter->free = 1;
	rewriter->prev = QSE_NULL;
	rewriter->next = ursd->free_rewriter;
	if (ursd->free_rewriter) ursd->free_rewriter->prev = rewriter;
	ursd->free_rewriter = rewriter;
	ursd->free_rewriter_count++;
}

static rewriter_t* dechain_rewriter_from_free_list (ursd_t* ursd, rewriter_t* rewriter)
{
	if (rewriter->next) rewriter->next->prev = rewriter->prev;

	if (rewriter == ursd->free_rewriter) ursd->free_rewriter = rewriter->next;
	else rewriter->prev->next = rewriter->next;

	rewriter->free = 0;
	ursd->free_rewriter_count--;
	return rewriter;
}

static void chain_rewriter_to_busy_list (ursd_t* ursd, rewriter_t* rewriter)
{
	rewriter->busy = 1;
	rewriter->prev = QSE_NULL;
	rewriter->next = ursd->busy_rewriter;
	if (ursd->busy_rewriter) ursd->busy_rewriter->prev = rewriter;
	ursd->busy_rewriter = rewriter;
}

static rewriter_t* dechain_rewriter_from_busy_list (ursd_t* ursd, rewriter_t* rewriter)
{
	if (rewriter->next) rewriter->next->prev = rewriter->prev;

	if (rewriter == ursd->busy_rewriter) ursd->busy_rewriter = rewriter->next;
	else rewriter->prev->next = rewriter->next;

	rewriter->busy = 0;
	return rewriter;
}

static void start_rewriter (ursd_t* ursd, rewriter_t* rewriter)
{
	QSE_ASSERT (rewriter->pio == QSE_NULL);
	
	rewriter->pio = qse_pio_open (
		ursd->mmgr, 0, ursd->cmdline, QSE_NULL, 
		QSE_PIO_WRITEIN | QSE_PIO_READOUT | QSE_PIO_ERRTONUL | 
		QSE_PIO_INNOBLOCK | QSE_PIO_OUTNOBLOCK
	);

	if (rewriter->pio)
	{
		if (insert_to_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_OUT), TYPE_PIO_OUT, rewriter->index) <= -1) 
		{
			/* error logging */
			qse_pio_kill (rewriter->pio);
			qse_pio_close (rewriter->pio);
			rewriter->pio = QSE_NULL;
		}
	}
}

static void stop_rewriter (ursd_t* ursd, rewriter_t* rewriter)
{
	if (rewriter->pio)
	{
		if (rewriter->pio_in_in_mux)
		{
			delete_from_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_IN), TYPE_PIO_IN, rewriter->index);
			rewriter->pio_in_in_mux = 0;
		}

		delete_from_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_OUT), TYPE_PIO_OUT, rewriter->index);

		qse_pio_kill (rewriter->pio);
		qse_pio_close (rewriter->pio);
		rewriter->pio = QSE_NULL;
	}
}

static void restart_rewriter (ursd_t* ursd, rewriter_t* rewriter)
{
	stop_rewriter (ursd, rewriter);
	start_rewriter (ursd, rewriter);
}

static void reset_rewriter_data (rewriter_t* rewriter)
{
	int index = rewriter->index;
	qse_pio_t* pio = rewriter->pio;

	qse_memset (rewriter, 0, QSE_SIZEOF(*rewriter));

	rewriter->index = index;
	rewriter->pio = pio;
}

static rewriter_t* get_free_rewriter (ursd_t* ursd)
{
	rewriter_t* rewriter;

	rewriter = ursd->free_rewriter;

	if (rewriter)
	{
		QSE_ASSERT (!rewriter->busy);
		QSE_ASSERT (!rewriter->faulty);
		QSE_ASSERT (!rewriter->pio_in_in_mux);

		if (!rewriter->pio) start_rewriter (ursd, rewriter); 

		dechain_rewriter_from_free_list (ursd, rewriter);
		reset_rewriter_data (rewriter);
	}

	return rewriter;
}

static void release_rewriter (ursd_t* ursd, rewriter_t* rewriter, int send_empty_response)
{
	if (send_empty_response)
	{
		urs_pkt_t* pkt = (urs_pkt_t*)rewriter->req.buf;
		pkt->hdr.pktlen = qse_ntoh16(QSE_SIZEOF(urs_hdr_t));
		sendto (ursd->sck, pkt, QSE_SIZEOF(urs_hdr_t), 0, (struct sockaddr*)&rewriter->req.from, rewriter->req.fromlen); 
		/* TOOD: error logging. if this fails, the client side should resend a request or just time out. */
	}

	if (rewriter->pio_in_in_mux)
	{
		QSE_ASSERT (rewriter->pio);
		delete_from_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_IN), TYPE_PIO_IN, rewriter->index);
		rewriter->pio_in_in_mux = 0;
	}

	if (rewriter->busy) dechain_rewriter_from_busy_list (ursd, rewriter);

	if (rewriter->faulty || !rewriter->pio) 
	{
		restart_rewriter (ursd, rewriter);

		/* NOTE: start may fail in restart_rewriter(), 
		 *       meaning rewrite->pio can still be null. */

		rewriter->faulty = 0;
	}

	chain_rewriter_to_free_list (ursd, rewriter);
}


static void seize_rewriter (ursd_t* ursd, rewriter_t* rewriter)
{
	/* call this function once the request from the socket has been
	 * fully written to the rewriter. */

	if (rewriter->pio_in_in_mux)
	{
		delete_from_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_IN), TYPE_PIO_IN, rewriter->index);
		rewriter->pio_in_in_mux = 0;
	}

	QSE_ASSERT (!rewriter->faulty);
	QSE_ASSERT (!rewriter->busy);

	chain_rewriter_to_busy_list (ursd, rewriter);
}

static int feed_rewriter (ursd_t* ursd, rewriter_t* rewriter)
{
	qse_ssize_t x;
	urs_pkt_t* pkt = (urs_pkt_t*)rewriter->req.buf;

	while (rewriter->req.urlpos < rewriter->req.urllen)
	{
		x = qse_pio_write (rewriter->pio, QSE_PIO_IN, &pkt->url[rewriter->req.urlpos], rewriter->req.urllen - rewriter->req.urlpos);
		if (x <= -1)
		{
			if (qse_pio_geterrnum(rewriter->pio) == QSE_PIO_EAGAIN)
			{
				if (rewriter->pio_in_in_mux || 
				    insert_to_mux (ursd->mux, qse_pio_gethnd(rewriter->pio, QSE_PIO_IN), TYPE_PIO_IN, rewriter->index) >= 0)
				{
					/* this is partial success. the request has not 
					   been passed to the rewriter in its entirety yet. */
					rewriter->pio_in_in_mux = 1;
					return 0; 
				}
			}

			/* reclaim the rewriter since it seems faulty */
			rewriter->faulty = 1;
			release_rewriter (ursd, rewriter, 1);
			return -1;
		}

		rewriter->req.urlpos += x;
	}

	/* full feeding is completed - the entire request has 
	 * been passed to the rewriter */
	seize_rewriter (ursd, rewriter);
	return 1; /* the full url has been passed to the rewriter */
}

static int enqueue_request (ursd_t* ursd, urs_pkt_t* pkt, const qse_skad_t* from, qse_sck_len_t fromlen)
{
	qse_uint16_t blkidx;
	qse_size_t memsize;
	xreq_t* xreq;

	/* TODO: if (ursd->xreq.count > MAX_QUEUE_SIZE) return -1; */

	blkidx = (pkt->hdr.pktlen - 1) / XREQ_BLOCK_SIZE; /* 0 based */
	if (blkidx < XREQ_MAX_BLOCKS)
	{
		xreq = ursd->xreq.free[blkidx];
		if (xreq) 
		{
			ursd->xreq.free[blkidx] = xreq->next;
			goto copy_packet;
		}
		memsize = (blkidx + 1) * XREQ_BLOCK_SIZE;
	}
	else
	{
		memsize = pkt->hdr.pktlen;
	}

	xreq = QSE_MMGR_ALLOC (ursd->mmgr, QSE_SIZEOF(*xreq) + memsize);
	if (xreq == QSE_NULL) return -1;

copy_packet:
	/* TODOO: xreq->timestamp */
	xreq->next = QSE_NULL;
	xreq->from = *from;
	xreq->fromlen = fromlen;
	xreq->pkt = (urs_pkt_t*)(xreq + 1);
	qse_memcpy (xreq->pkt, pkt, pkt->hdr.pktlen);

	if (ursd->xreq.count > 0)
	{
		ursd->xreq.tail->next = xreq;
		ursd->xreq.tail = xreq;
	}
	else
	{
		ursd->xreq.head = xreq;
		ursd->xreq.tail = xreq;
	}

	ursd->xreq.count++;
	return 0;
}

static xreq_t* dequeue_request (ursd_t* ursd)
{
	xreq_t* xreq = QSE_NULL;

	if (ursd->xreq.count > 0)
	{
		xreq = ursd->xreq.head;

		ursd->xreq.head = xreq->next;
		if (ursd->xreq.count == 1)
			ursd->xreq.tail = ursd->xreq.head;

		ursd->xreq.count--;
	}

	return xreq;
}

static void release_request (ursd_t* ursd, xreq_t* xreq)
{
	qse_uint16_t blkidx;

	blkidx = (xreq->pkt->hdr.pktlen - 1) / XREQ_BLOCK_SIZE;
	if (blkidx < XREQ_MAX_BLOCKS)
	{
		xreq->next = ursd->xreq.free[blkidx];
		ursd->xreq.free[blkidx] = xreq;
	}
	else
	{
		QSE_MMGR_FREE (ursd->mmgr, xreq);
	}
}

static void handle_pending_requests (ursd_t* ursd)
{
	rewriter_t* rewriter;
	xreq_t* xreq;

	while (ursd->xreq.count > 0)
	{
		rewriter = get_free_rewriter (ursd);
		if (rewriter)
		{
	qse_printf (QSE_T("HANDLING PENDING REQUEST.... %d\n"), (int)ursd->xreq.count);
			xreq = dequeue_request (ursd);
			QSE_ASSERT (xreq);

			rewriter->req.fromlen = xreq->fromlen;
			rewriter->req.from = xreq->from;
			qse_memcpy (rewriter->req.buf, xreq->pkt, xreq->pkt->hdr.pktlen);
			rewriter->req.buf[xreq->pkt->hdr.pktlen] = QSE_MT('\n'); /* put a new line at the end */
			rewriter->req.urlpos = 0;
			rewriter->req.urllen = xreq->pkt->hdr.pktlen - QSE_SIZEOF(urs_hdr_t) + 1;  /* +1 for '\n' */

			release_request (ursd, xreq);

			if (rewriter->pio) feed_rewriter (ursd, rewriter);
			else release_rewriter (ursd, rewriter, 1);
		}
		else
		{
	qse_printf (QSE_T("NO REWRITER AVAILABLE FOR PENDING REQUEST....%d\n"), (int)ursd->xreq.count);
			break;
		}
	}
}

static void receive_request_from_socket (ursd_t* ursd, const qse_mux_evt_t* evt)
{
	qse_ssize_t x;
	urs_pkt_t* pkt;
	rewriter_t* rewriter;

	rewriter = get_free_rewriter (ursd);
	if (rewriter)
	{
		rewriter->req.fromlen = QSE_SIZEOF(rewriter->req.from);
		x = recvfrom (evt->hnd, rewriter->req.buf, QSE_SIZEOF(rewriter->req.buf) - 1, 0, (struct sockaddr*)&rewriter->req.from, &rewriter->req.fromlen);
		/*TODO: improve error handling */
		if (x < QSE_SIZEOF(urs_hdr_t)) 
		{
			/* TODO: message logging */
			return;
		}

		pkt = (urs_pkt_t*)rewriter->req.buf;
		pkt->hdr.pktlen = qse_ntoh16(pkt->hdr.pktlen); /* change the byte order */
		if (pkt->hdr.pktlen != x)
		{
			/* TOOD: message logging */
			return;
		}

		if (rewriter->pio)
		{
			rewriter->req.buf[x] = QSE_MT('\n'); /* put a new line at the end */
			rewriter->req.urlpos = 0;
			rewriter->req.urllen = pkt->hdr.pktlen - QSE_SIZEOF(urs_hdr_t) + 1; /* +1 for '\n' */

qse_printf (QSE_T("%d [[[%.*hs]]]\n"), (int)x, (int)rewriter->req.urllen, pkt->url);
			feed_rewriter (ursd, rewriter); 
		}
		else
		{
			/* the actual rewriter is defunct. */
			/* TODO: error logging */
qse_printf (QSE_T("rewriter->pio is NULL. ....\n"));
			release_rewriter (ursd, rewriter, 1);
		}
	}
	else
	{
		qse_sck_len_t fromlen;
		qse_skad_t from;
		urs_pkt_t* pkt;

		fromlen = QSE_SIZEOF(from);
		x = recvfrom (evt->hnd, ursd->xreq.buf, QSE_SIZEOF(ursd->xreq.buf) - 1, 0, (struct sockaddr*)&from, &fromlen);

		if (x < QSE_SIZEOF(urs_hdr_t)) 
		{
			/* TODO: message logging */
			return;
		}

		pkt = (urs_pkt_t*)ursd->xreq.buf;
		pkt->hdr.pktlen = qse_ntoh16(pkt->hdr.pktlen); /* change the byte order */
		if (pkt->hdr.pktlen != x)
		{
			/* TOOD: message logging */
			return;
		}

		if (enqueue_request (ursd, pkt, &from, fromlen) <= -1)
		{
			/* TODO: error logging for failed enqueuing */

			/* if enqueuing fails, send an empty response */
			pkt->hdr.rcode = qse_hton16(URS_RCODE_ERROR);
			pkt->hdr.pktlen = qse_hton16(QSE_SIZEOF(urs_hdr_t));
			sendto (evt->hnd, pkt, QSE_SIZEOF(urs_hdr_t), 0, (struct sockaddr*)&from, fromlen);
			/* TODO: error logging for sendto failure */
		}
	}
}

static void dispatch_mux_event (qse_mux_t* mux, const qse_mux_evt_t* evt)
{
	mux_xtn_t* mux_xtn;
	int type, index;

	mux_xtn = (mux_xtn_t*)qse_mux_getxtn(mux);

	type = GET_TYPE_FROM_MUX_DATA((qse_uintptr_t)evt->data);
	index = GET_INDEX_FROM_MUX_DATA((qse_uintptr_t)evt->data);

	switch (type)
	{
		case TYPE_SOCKET:
		{
			receive_request_from_socket (mux_xtn->ursd, evt);
			break;
		}

		case TYPE_PIO_OUT:
		{
			/* the rewriter has produced some data */

			qse_ssize_t x;
			urs_pkt_t* pkt;
			qse_size_t maxoutlen;

			rewriter_t* rewriter = &mux_xtn->ursd->rewriters[index];

			if (rewriter->busy)
			{
				pkt = (urs_pkt_t*)rewriter->req.buf;
				maxoutlen = QSE_SIZEOF(rewriter->req.buf) - QSE_SIZEOF(urs_hdr_t);

				x = qse_pio_read (rewriter->pio, QSE_PIO_OUT, &pkt->url[rewriter->req.outlen], maxoutlen - rewriter->req.outlen);
				if (x <= 0)
				{
					/* read failure or end of input */
					rewriter->faulty = 1;
					release_rewriter (mux_xtn->ursd, rewriter, 1);
					/* TODO: error logging */
				}
				else
				{
					rewriter->req.outlen += x;

qse_printf (QSE_T("READ %d, %d bytes from pipes [%.*hs]\n"), (int)x, (int)rewriter->req.outlen, (int)rewriter->req.outlen, pkt->url);

					if (rewriter->req.outlen > maxoutlen)
					{
						/* the rewriter returns too long a result */
						rewriter->faulty = 1;
						release_rewriter (mux_xtn->ursd, rewriter, 1);
						/* TODO: error logging */
					}
					else if (pkt->url[rewriter->req.outlen - 1] == QSE_MT('\n'))
					{
						/* the last byte is a new line. i don't really care about
						 * new lines in the middle of data. the rewriter must 
						 * keep to the protocol. */

						/* add up the header size. -1 to exclude '\n' */
						rewriter->req.outlen += QSE_SIZEOF(urs_hdr_t) - 1;

						pkt->hdr.pktlen = qse_hton16(rewriter->req.outlen); /* change the byte order */
						sendto (mux_xtn->ursd->sck, pkt, rewriter->req.outlen, 0, (struct sockaddr*)&rewriter->req.from, rewriter->req.fromlen);
						/* TODO: error logging */
						/* sendto() to the socket can be blocking. if the socket side is too busy, there's no reason for rewriter to be as busy.
						 * it can wait a while. think about it. */

						release_rewriter (mux_xtn->ursd, rewriter, 0);
					}
				}

/* TODO: is the complete output is not received within time, some actions must be taken. timer based... rewrite timeout */
			}
			else 
			{
				/* something is wrong. if the rewriter process writes something 
				 * while it's not given the full input. reclaim it */
				rewriter->faulty = 1;
				release_rewriter (mux_xtn->ursd, rewriter, 1);
			}

			break;
		}

		case TYPE_PIO_IN:
		{
			/* the pipe to the rewriter is writable. 
			 * pass the leftover to the rewriter */

			rewriter_t* rewriter = &mux_xtn->ursd->rewriters[index];

			QSE_ASSERT (rewriter->pio_in_in_mux);
			QSE_ASSERT (!rewriter->free && !rewriter->busy);

			feed_rewriter (mux_xtn->ursd, rewriter);
			break;
		}
	}
}

static int init_ursd (ursd_t* ursd, int npios, const qse_char_t* cmdline, const qse_char_t* bindaddr)
{
	qse_size_t i;
	qse_nwad_t bindnwad;
	mux_xtn_t* mux_xtn;

	if (npios <= 0) npios = 1;

	qse_memset (ursd, 0, sizeof(*ursd));
	ursd->mmgr = QSE_MMGR_GETDFL();
	
	ursd->cmdline = qse_strdup (cmdline, ursd->mmgr);
	if (ursd->cmdline == QSE_NULL)
	{
		qse_fprintf (QSE_STDERR, QSE_T("cannot duplicate cmdline\n"));
		goto oops;
	}

	ursd->mux = qse_mux_open (ursd->mmgr, QSE_SIZEOF(mux_xtn_t), dispatch_mux_event, 100, QSE_NULL);
	if (ursd->mux == QSE_NULL)
	{
		qse_fprintf (QSE_STDERR, QSE_T("cannot create a multiplexer\n"));
		goto oops;
	}

	if (qse_strtonwad (bindaddr, &bindnwad) <= -1)
	{
		qse_fprintf (QSE_STDERR, QSE_T("invalid binding address\n"));
		goto oops;
	}

	ursd->sck = open_server_socket (/*IPPROTO_SCTP*/IPPROTO_UDP, &bindnwad);
	if (ursd->sck == QSE_INVALID_SCKHND) goto oops;

	ursd->rewriters = QSE_MMGR_ALLOC (ursd->mmgr, npios * QSE_SIZEOF(rewriter_t));
	if (ursd->rewriters == QSE_NULL) 
	{
		qse_fprintf (QSE_STDERR, QSE_T("cannot allocate rewriters\n"));
		goto oops;
	}
	qse_memset (ursd->rewriters, 0, npios * QSE_SIZEOF(rewriter_t));

	for (i = 0; i < npios; i++) 
	{
		ursd->rewriters[i].index = i;
		start_rewriter (ursd, &ursd->rewriters[i]);
		release_rewriter (ursd, &ursd->rewriters[i], 0);
	}

	if (insert_to_mux (ursd->mux, ursd->sck, TYPE_SOCKET, 0) <= -1) 
	{
		qse_fprintf (QSE_STDERR, QSE_T("cannot add socket to multiplexer\n"));
		goto oops;
	}
	
	ursd->total_rewriter_count = npios;

	mux_xtn = qse_mux_getxtn (ursd->mux);
	mux_xtn->ursd = ursd;
	return 0;

oops:
	if (ursd->rewriters) 
	{
		for (i = 0; i < npios; i++) stop_rewriter (ursd, &ursd->rewriters[i]);
		QSE_MMGR_FREE (ursd->mmgr, ursd->rewriters);
	}
	if (qse_is_sck_valid(ursd->sck)) qse_close_sck (ursd->sck);
	if (ursd->mux) qse_mux_close (ursd->mux);
	if (ursd->cmdline) QSE_MMGR_FREE(ursd->mmgr, ursd->cmdline);

	return -1;
}

static void fini_ursd (ursd_t* ursd)
{
	qse_size_t i;
	xreq_t* xreq;

	for (i = 0; i < ursd->total_rewriter_count; i++) 
		stop_rewriter (ursd, &ursd->rewriters[i]);

	QSE_MMGR_FREE (ursd->mmgr, ursd->rewriters);

	delete_from_mux (ursd->mux, ursd->sck, TYPE_SOCKET, 0);
	qse_close_sck (ursd->sck);
	qse_mux_close (ursd->mux);
	QSE_MMGR_FREE (ursd->mmgr, ursd->cmdline);


	/* destroy the request queue */
	xreq = ursd->xreq.head;
	while (xreq)
	{
		xreq_t* next = xreq->next;
		QSE_MMGR_FREE (ursd->mmgr, xreq);
		xreq = next;
	}

	for (i = 0; i < QSE_COUNTOF(ursd->xreq.free); i++)
	{
		while (ursd->xreq.free[i])
		{
			xreq = ursd->xreq.free[i];
			ursd->xreq.free[i] = xreq->next;
			QSE_MMGR_FREE (ursd->mmgr, xreq);
		}
	}
}

static int g_stop_requested = 0;
static void handle_signal (int sig)
{
	switch (sig)
	{
		case SIGINT:
		case SIGTERM:
	#if defined(SIGHUP)
		case SIGHUP:
	#endif
			g_stop_requested = 1;
			break;
	}
}

static void print_usage (qse_char_t* argv0)
{
	qse_fprintf (QSE_STDERR, QSE_T("Usage: %s rewriter-path rewriter-count binding-address\n"), qse_basename(argv0));
}

static int httpd_main (int argc, qse_char_t* argv[])
{
	ursd_t ursd;
	int ursd_inited = 0;


	if (argc < 4)
	{
		print_usage (argv[0]);
		return -1;
	}

	signal (SIGINT, handle_signal);
	signal (SIGTERM, handle_signal);
#if defined(SIGHUP)
	signal (SIGHUP, handle_signal);
#endif
#if defined(SIGPIPE)
	signal (SIGPIPE, SIG_IGN);
#endif

	if (init_ursd (&ursd, qse_strtoi(argv[2], 10, QSE_NULL), argv[1], argv[3]) <= -1) goto oops;
	ursd_inited = 1;

	while (!g_stop_requested)
	{
		qse_ntime_t tmout;
		qse_clear_ntime (&tmout);

		/* if there are pending requests, use timeout of 0.
		 * this way, multiplexer events and pending requests can be
		 * handled together. note if there's no free rewriter while
		 * there's a pending request, the request can't be performed.
		 * the free rewriter check is simpler than the actual check
		 * in get_free_rewriter(). it's also possible that no free
		 * rewriter is available after qse_mux_poll(). this timeout
		 * calculation is on the best-effort basis. */
		if (ursd.xreq.count <= 0 || !ursd.free_rewriter) tmout.sec += 1;
/* TODO: add timer also... and consider that in calculating in tmout */

		qse_mux_poll (ursd.mux, &tmout);

		handle_pending_requests (&ursd);
	}

	fini_ursd (&ursd);
	return 0;

oops:
	if (ursd_inited) fini_ursd (&ursd);
	return -1;
}

int qse_main (int argc, qse_achar_t* argv[])
{
	int ret;

#if defined(_WIN32)
	char locale[100];
	UINT codepage;
	WSADATA wsadata;
#else
	/* nothing */
#endif

#if defined(_WIN32)

	codepage = GetConsoleOutputCP();
	if (codepage == CP_UTF8)
	{
		/*SetConsoleOUtputCP (CP_UTF8);*/
		qse_setdflcmgrbyid (QSE_CMGR_UTF8);
	}
	else
	{
		/* .codepage */
		qse_fmtuintmaxtombs (locale, QSE_COUNTOF(locale),
			codepage, 10, -1, QSE_MT('\0'), QSE_MT("."));
		setlocale (LC_ALL, locale);
		/*qse_setdflcmgrbyid (QSE_CMGR_SLMB);*/
	}

	if (WSAStartup (MAKEWORD(2,0), &wsadata) != 0)
	{
		qse_fprintf (QSE_STDERR, QSE_T("failed to start up winsock\n"));
		return -1;
	}

#else
	setlocale (LC_ALL, "");
	/*qse_setdflcmgrbyid (QSE_CMGR_SLMB);*/
#endif

	qse_open_stdsios ();
	ret = qse_run_main (argc, argv, httpd_main);
	qse_close_stdsios ();


#if defined(_WIN32)
	WSACleanup ();
#endif

	return ret;
}