/*
 * Copyright (C) 1994-2018 Altair Engineering, Inc.
 * For more information, contact Altair at www.altair.com.
 *
 * This file is part of the PBS Professional ("PBS Pro") software.
 *
 * Open Source License Information:
 *
 * PBS Pro is free software. You can redistribute it and/or modify it under the
 * terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option) any
 * later version.
 *
 * PBS Pro is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * Commercial License Information:
 *
 * For a copy of the commercial license terms and conditions,
 * go to: (http://www.pbspro.com/UserArea/agreement.html)
 * or contact the Altair Legal Department.
 *
 * Altair’s dual-license business model allows companies, individuals, and
 * organizations to create proprietary derivative works of PBS Pro and
 * distribute them - whether embedded or bundled with other software -
 * under a commercial license agreement.
 *
 * Use of Altair’s trademarks, including but not limited to "PBS™",
 * "PBS Professional®", and "PBS Pro™" and Altair’s logos is subject to Altair's
 * trademark licensing policies.
 *
 */

/**
 * @file tcp_dis_win.c
 *
 * @brief
 *	TCP DIS related functions (used by both the client side and server side
 *	calls)
 *
 * @warning
 *
 * Functions in this file are NOT thread safe. If multiple threads
 * call different functions simultaneously with the same fd, the
 * it will corrupt global data.\n
 *
 * Thread safety has to be ensured at the caller level.\n
 *
 * In this case the the client connection layer is synchronized on
 * each connection, so multiple theads calling these routines from the IFL is
 * okay. The connection lock (ch_mutex) used by upper level callers
 * (from IFL API).\n
 *
 * @see tcp_get_readbuf\n tcp_get_writebuf
 * These routines are synchronized using the tcp lock (global lock) so that
 * access to the global array of read/write buffers (struct tcp_chan **)
 * is synchronized.
 *
 * Calls to these routines from the daemons are not synchronized currently as
 * the daemons are single threaded.
 *
 */


#include <pbs_config.h>   /* the master config generated by configure */

#include <errno.h>
#include <stdio.h>

#include <sys/types.h>
#ifdef WIN32
#include <time.h>
#include <io.h>
#include "pbs_ifl.h"
#include "net_connect.h"
#include "log.h"
#include <assert.h>
#else
#include <unistd.h>
#include <sys/time.h>
#endif

#include <stdlib.h>
#include "libpbs.h"
#include "rpp.h"

#if defined(FD_SET_IN_SYS_SELECT_H)
#  include <sys/select.h>
#endif

#include "dis.h"
#include "dis_init.h"

#define THE_BUF_SIZE 1024

struct tcpdisbuf {
	size_t	tdis_lead;
	size_t	tdis_trail;
	size_t	tdis_eod;
	size_t	tdis_bufsize;
	char	*tdis_thebuf;
};

struct	tcp_chan {
	struct	tcpdisbuf	readbuf;
	struct	tcpdisbuf	writebuf;
};

#ifdef WIN32
struct conlink {
	int		c_fd;
	struct tcp_chan	c_tcp;
	struct conlink  *c_link;
};
#define TCP_HASH 0x1f

static struct conlink *tcparray[TCP_HASH+1];
#else
static	struct	tcp_chan	**tcparray = NULL;
static	int			tcparraymax = 0;
#endif


/* windows 2000 needs prototypes */
#ifdef WIN32
static void DIS_tcp_clear(struct tcpdisbuf *tp);
static void DIS_delete_tcp_chan(int fd);

static int tcp_getc(int fd);
static int tcp_puts(int fd, const char *str, size_t ct);
static int tcp_gets(int fd, char *str, size_t ct);
static int tcp_rskip(int fd, size_t ct);
static int tcp_rcommit(int fd, int commit_flag);
static int tcp_wcommit(int fd, int commit_flag);

/**
 * @brief
 *	-initialize the tcparray.
 *
 */
void
DIS_tcparray_init(void)
{
	int	i;

	for (i=0; i <=TCP_HASH; i++) {
		tcparray[i] = NULL;
	}
}

static struct tcp_chan *DIS_find_tcp_chan(int fd)
{
	int	hent;
	struct  conlink *found = NULL;

	hent = fd & TCP_HASH;

	found = tcparray[hent];

	while (found) {
		if (found->c_fd == fd)
			break;
		found = found->c_link;
	}
	if (found == NULL)
		return NULL;

	return &found->c_tcp;
}

/**
 * @brief
 *	deletes the tcp channel.
 *
 * @param[in] fd - file descriptor which to be closed
 *
 */
static void
DIS_delete_tcp_chan(int fd)
{
	int	hent;
	struct  conlink *found = NULL;
	struct  conlink *found_before = NULL;

	hent = fd & TCP_HASH;

	found = tcparray[hent];
	found_before = NULL;
	while (found) {
		if (found->c_fd == fd) {
			if (found_before == NULL) {
				tcparray[hent] = found->c_link;
			} else {
				found_before->c_link = found->c_link;
			}
			if (found->c_tcp.readbuf.tdis_thebuf)
				free(found->c_tcp.readbuf.tdis_thebuf);

			if (found->c_tcp.writebuf.tdis_thebuf)
				free(found->c_tcp.writebuf.tdis_thebuf);
			free(found);
			return;
		}
		found_before = found;
		found = found->c_link;
	}

}

/**
 * @brief
 *	ctreate tcp channel.
 *
 * @param[in] fd - file descriptor
 *
 * @return	structure handle
 * @retval	pointer to tcp channel	success
 * @retval	NULL			error
 *
 */
static struct tcp_chan *DIS_create_tcp_chan(int fd)
{
	struct conlink *cl;
	int hent;

	DIS_delete_tcp_chan(fd);

	cl = malloc(sizeof(struct conlink));
	assert(cl != NULL);

	cl->c_fd = fd;
	cl->c_tcp.readbuf.tdis_thebuf = malloc(THE_BUF_SIZE);
	if (cl->c_tcp.readbuf.tdis_thebuf == NULL) {
		(void)free(cl);
		return NULL;
	}
	cl->c_tcp.readbuf.tdis_bufsize = THE_BUF_SIZE;
	cl->c_tcp.writebuf.tdis_thebuf = malloc(THE_BUF_SIZE);
	if (cl->c_tcp.writebuf.tdis_thebuf == NULL) {
		(void)free(cl->c_tcp.readbuf.tdis_thebuf);
		(void)free(cl);
		return NULL;
	}
	cl->c_tcp.writebuf.tdis_bufsize = THE_BUF_SIZE;
	cl->c_link = NULL;

	DIS_tcp_clear(&cl->c_tcp.readbuf);
	DIS_tcp_clear(&cl->c_tcp.writebuf);

	hent = fd & TCP_HASH;

	cl->c_link = tcparray[hent];
	tcparray[hent] = cl;

	return &cl->c_tcp;
}
#endif


/**
 * @brief
 *	Synchronize access to readbuf location
 *
 * @par Functionality:
 *	This routine is used by all other (read) routines to synchronously
 *	retrieve the location of the tcp read buffer for the supplied fd. It
 *	uses the tcp lock (defined in pbs_client_thread.c) to synchronize
 *	access to the array of tcp_chans.\n
 *
 *	The reason that this has to be synchronized, is that the global array of
 *	read/write buffers (struct tcp_chan **) could be realloc'd by routine
 *	@see DIS_tcp_setup - resulting in the whole array to get relocated
 *	elsewhere in memory. However, while the whole array can be relocated
 *	the individual pointers in the array continue to point to the same
 *	address of the read/write buffers. Thus only the point of access to the
 *	global array is synchronized, not the actual read/writes using the
 *	retrieved read/write buffers.
 *
 * @param[in] fd - The file handle for the socket for which the buffer is to be
 *		   accessed
 *
 * @retval - address of the read buffer to use for readers
 *
 * @par Side-effects:
 *	Uses assert to ensure that the buffer address retrieved is not NULL
 */
struct tcpdisbuf * tcp_get_readbuf(int fd)
{
	struct	tcpdisbuf	*tp;
	int rc;

	rc = pbs_client_thread_lock_tcp();
	assert(rc == 0);

#ifdef WIN32
	tp = &(DIS_find_tcp_chan(fd)->readbuf);
#else
	tp = &tcparray[fd]->readbuf;
#endif
	rc = pbs_client_thread_unlock_tcp();
	assert(rc == 0);

	assert(tp != NULL);
	return (tp);
}

/**
 * @brief
 *	Synchronize access to writebuf location
 *
 * @par Functionality:
 *	This routine is used by all other (write) routines to synchronously
 *	retrieve the location of the tcp write buffer for the supplied fd. It
 *	uses the tcp lock (defined in pbs_client_thread.c) to synchronize
 *	access to the array of tcp_chans.\n
 *
 *	The reason that this has to be synchronized, is that the global array of
 *	read/write buffers (struct tcp_chan **) could be realloc'd by routine
 *	@see DIS_tcp_setup - resulting in the whole array to get relocated
 *	elsewhere in memory. However, while the whole array can be relocated
 *	the individual pointers in the array continue to point to the same
 *	address of the read/write buffers. Thus only the point of access to the
 *	global array is synchronized, not the actual read/writes using the
 *	retrieved read/write buffers.
 *
 * @param[in] fd - The file handle for the socket for which the buffer is to be
 *		   accessed
 *
 * @retval - address of the write buffer to use for writers
 *
 * @par Side-effects:
 *	Uses assert to ensure that the buffer address retrieved is not NULL
 */
struct tcpdisbuf * tcp_get_writebuf(int fd)
{
	struct	tcpdisbuf	*tp;
	int rc;

	rc = pbs_client_thread_lock_tcp();
	assert(rc == 0);

#ifdef WIN32
	tp = &(DIS_find_tcp_chan(fd)->writebuf);
#else
	tp = &tcparray[fd]->writebuf;
#endif

	rc = pbs_client_thread_unlock_tcp();
	assert(rc == 0);

	assert(tp != NULL);
	return (tp);
}


/**
 * @brief
 * 	-tcp_pack_buff - pack existing data into front of buffer
 *
 * @par	Functionality:
 *	Moves "uncommited" data to front of buffer and adjusts pointers.
 *	Does a character by character move since data may over lap.
 *
 * @param[in] tp - pointer to tcpdisbuf
 *
 * @return	Void
 *
 */

static void
tcp_pack_buff(struct tcpdisbuf *tp)
{
	size_t amt;
	size_t start;
	size_t i;

	start = tp->tdis_trail;
	if (start != 0) {
		amt  = tp->tdis_eod - start;
		for (i=0; i<amt; ++i) {
			*(tp->tdis_thebuf + i) =
				*(tp->tdis_thebuf + i + start);
		}
		tp->tdis_lead  -= start;
		tp->tdis_trail -= start;
		tp->tdis_eod   -= start;
	}
}

/**
 * @brief
 * 	tcp_read - read data from tcp stream to "fill" the buffer
 *	Update the various buffer pointers.
 *
 * @param[in] fd - file descriptor
 *
 * @return	int
 * @retval	>0 	number of characters read
 * @retval	0 	if EOD (no data currently avalable)
 * @retval	-1 	if error
 * @retval	-2 	if EOF (stream closed)
 */

static int
tcp_readi(int fd)
{
	int i;
	fd_set	readset;
	struct	timeval timeout;
	struct	tcpdisbuf	*tp;

	tp = tcp_get_readbuf(fd);

	tcp_pack_buff(tp);	/* compact uncommitted data */
	if ((tp->tdis_bufsize - tp->tdis_eod) < 20) {	/* need room */
		tp->tdis_bufsize += THE_BUF_SIZE;
		tp->tdis_thebuf = (char *)realloc(tp->tdis_thebuf,
			sizeof(char)*tp->tdis_bufsize);
	}

	/*
	 * we don't want to be locked out by an attack on the port to
	 * deny service, so we time out the read, the network had better
	 * deliver promptly
	 */
	do {
		timeout.tv_sec = (long) pbs_tcp_timeout;
		timeout.tv_usec = 0;

		FD_ZERO(&readset);
		FD_SET((unsigned int)fd, &readset);
		i = select(FD_SETSIZE, &readset, NULL,
			NULL, &timeout);
		if (pbs_tcp_interrupt)
			break;
#ifdef WIN32
	} while ((i == -1) && (errno == WSAEINTR));
#else
	} while ((i == -1) && (errno == EINTR));
#endif

	if ((i == 0) || (i < 0))
		return i;

#ifdef WIN32
	/* need CS_recv() interface when we extend to windows */

	while ((i = recv(fd, &tp->tdis_thebuf[tp->tdis_eod],
		(int) (tp->tdis_bufsize - tp->tdis_eod), 0)) == -1) {
		errno = WSAGetLastError();
		if (errno != WSAEINTR) {
			if (errno == WSAECONNRESET) {
				i = 0;	/* treat like no data for winsock */
				/* will return this if remote */
				/* connection prematurely closed */
			}

			break;
		}
#else
	while ((i = CS_read(fd, &tp->tdis_thebuf[tp->tdis_eod],
		tp->tdis_bufsize - tp->tdis_eod)) == CS_IO_FAIL) {

		if (errno != EINTR)
			break;
#endif
	}
	if (i > 0)
		tp->tdis_eod += i;
	return ((i == 0) ? -2 : i);
}

/**
 * @brief
 * 	-DIS_wflush - Wrapper function to do a tcp or TPP write buffer
 *
 * @param[in] sock - socket descriptor
 * @param[in] rpp - indication whether to use rpp or not
 *
 * @par	Functionality:
 * 	calls DIS_tcp_wflush or rpp_flush based on input parameter rpp
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 */
int
DIS_wflush(int sock, int rpp)
{
	if (rpp)
		return (rpp_flush(sock));
	else
		return (DIS_tcp_wflush(sock));
}

/**
 * @brief
 * 	DIS_tcp_wflush - flush tcp/dis write buffer
 *
 * @par	Functionality:
 *	Writes "committed" data in buffer to file discriptor,
 *	packs remaining data (if any), resets pointers
 *
 * @param[in] fd = file descriptor
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 *
 */

int
DIS_tcp_wflush(int fd)
{
	size_t	ct;
	int	i;
	char	*pb;
	struct	tcpdisbuf	*tp;

	pbs_tcp_errno = 0;
	tp = tcp_get_writebuf(fd);
	pb = tp->tdis_thebuf;

	ct = tp->tdis_trail;
	if (ct == 0)
		return 0;

#ifdef WIN32
	/* need CS_send() interface when we extend to windows */

	/*
	 printf("about to send pb=%s ct=%d\n", pb, ct);
	 */
	while ((i = send(fd, pb, (int) ct, 0)) != (int)ct) {
		errno = WSAGetLastError();
		if (i == -1) {
			if (errno != WSAEINTR) {
				pbs_tcp_errno = errno;
				return (-1);
			} else
				continue;
		}
#else
	while ((i = CS_write(fd, pb, ct)) != ct) {
		if (i == CS_IO_FAIL) {
			if (errno != EINTR) {
				pbs_tcp_errno = errno;
				return (-1);
			} else
				continue;
		}
#endif
		ct -= i;
		pb += i;
	}
	/*
	 printf("SENT pb=%s ct=%d\n", pb, ct);
	 */

	tp->tdis_eod = tp->tdis_lead;
	tcp_pack_buff(tp);
	return 0;
}

/**
 * @brief
 * 	DIS_buf_clear - reset tpc/dis buffer to empty
 *
 * @param[in] tp - pointer to buf which to be cleared
 *
 * @return	Void
 *
 */

static void
DIS_tcp_clear(struct tcpdisbuf *tp)
{
	tp->tdis_lead  = 0;
	tp->tdis_trail = 0;
	tp->tdis_eod   = 0;
}

/**
 * @brief
 *	-reset setup for tcp
 *
 * @param[in] fd - file descriptor
 * @param[in] i - indicates reads or write
 *
 * @return	Void
 *
 */
void
DIS_tcp_reset(int fd, int i)
{
	DIS_tcp_clear(i==0 ? tcp_get_readbuf(fd) : tcp_get_writebuf(fd));
}

/**
 * @brief
 * 	tcp_rskip - tcp/dis suport routine to skip over data in read buffer
 *
 * @param[in] fd - file descriptor
 * @param[in] ct - number of chars to skip
 *
 * @return	int
 * @retval	number of characters skipped
 *
 */

static int
tcp_rskip(int fd, size_t ct)
{
	struct	tcpdisbuf	*tp;

	tp = tcp_get_readbuf(fd);
	if (tp->tdis_lead - tp->tdis_eod < ct)
		ct = tp->tdis_lead - tp->tdis_eod;
	tp->tdis_lead += ct;
	return (int)ct;
}

/**
 * @brief
 * 	-tcp_getc - tcp/dis support routine to get next character from read buffer
 *
 * @param[in] fd - file descriptor
 *
 * @return	int
 * @retval	>0 	number of characters read
 * @retval	-1 	if EOD or error
 * @retval	-2 	if EOF (stream closed)
 */

static int
tcp_getc(int fd)
{
	int	x;
	struct	tcpdisbuf	*tp;

	tp = tcp_get_readbuf(fd);
	if (tp->tdis_lead >= tp->tdis_eod) {
		/* not enought data, try to get more */
		x = tcp_readi(fd);
		if (x <= 0)
			return ((x == -2) ? -2 : -1);	/* Error or EOF */
	}
	return ((int)tp->tdis_thebuf[tp->tdis_lead++]);
}

/**
 * @brief
 * 	-tcp_gets - tcp/dis support routine to get a string from read buffer
 *
 * @param[in] fd - file descriptor
 * @param[out] ct - num of chars read
 * @param[out] str - string read from buffer
 *
 * @return	int
 * @retval	>0 	number of characters read
 * @retval	 0 	if EOD (no data currently avalable)
 * @retval	-1	if error
 * @retval	-2 	if EOF (stream closed)
 */

static int
tcp_gets(int fd, char *str, size_t ct)
{
	int	x;
	struct	tcpdisbuf	*tp;

	tp = tcp_get_readbuf(fd);
	while (tp->tdis_eod - tp->tdis_lead < ct) {
		/* not enought data, try to get more */
		x = tcp_readi(fd);
		if (x <= 0)
			return x;	/* Error or EOF */
	}
	(void)memcpy(str, &tp->tdis_thebuf[tp->tdis_lead], ct);
	tp->tdis_lead += ct;
	return (int)ct;
}

/**
 * @brief
 *	-tcp_puts - tcp/dis support routine to put a counted string of characters
 *	into the write buffer.
 *
 * @param[in] fd -file descriptor
 * @param[in] str - string to be written
 * @param[out] ct - number of chars written
 *
 * @return	int
 * @retval	>= 0	the number of characters placed
 * @retval	-1 	if error
 */

static int
tcp_puts(int fd, const char *str, size_t ct)
{
	struct	tcpdisbuf	*tp;

	tp = tcp_get_writebuf(fd);
	if ((tp->tdis_bufsize - tp->tdis_lead) < ct) {
		/* not enough room, try to flush committed data */
		if (DIS_tcp_wflush(fd) < 0)
			return -1;		/* error */

		if ((tp->tdis_bufsize - tp->tdis_lead) < ct) {	/* add room */
			size_t	ru = (ct + tp->tdis_lead) / THE_BUF_SIZE;

			tp->tdis_bufsize = (ru + 1) * THE_BUF_SIZE;
			tp->tdis_thebuf = (char *)realloc(tp->tdis_thebuf,
				sizeof(char)*tp->tdis_bufsize);
		}
	}
	(void)memcpy(&tp->tdis_thebuf[tp->tdis_lead], str, ct);
	tp->tdis_lead += ct;
	return (int) ct;
}

/**
 * @brief
 * 	tcp_rcommit - tcp/dis support routine to commit/uncommit read data
 *
 * @param[in] fd - file descriptor
 * @param[in] flag - indication to commit or not
 *
 * @return      int
 * @retval      0       success
 */
static int
tcp_rcommit(int fd, int commit_flag)
{
	struct	tcpdisbuf	*tp;

	tp = tcp_get_readbuf(fd);
	if (commit_flag) {
		/* commit by moving trailing up */
		tp->tdis_trail = tp->tdis_lead;
	} else {
		/* uncommit by moving leading back */
		tp->tdis_lead = tp->tdis_trail;
	}
	return 0;
}

/**
 * @brief
 * 	-tcp_wcommit - tcp/dis support routine to commit/uncommit write data
 *
 * @param[in] fd - file descriptor
 * @param[in] flag - indication to commit or not
 *
 * @return      int
 * @retval      0       success
 */

static int
tcp_wcommit(int fd, int commit_flag)
{
	struct	tcpdisbuf	*tp;

	tp = tcp_get_writebuf(fd);
	if (commit_flag) {
		/* commit by moving trailing up */
		tp->tdis_trail = tp->tdis_lead;
	} else {
		/* uncommit by moving leading back */
		tp->tdis_lead = tp->tdis_trail;
	}
	return 0;
}

/**
 * @brief
 * 	tcp functions
 */
void
DIS_tcp_funcs()
{
	if (dis_getc != tcp_getc) {
		dis_getc = tcp_getc;
		dis_puts = tcp_puts;
		dis_gets = tcp_gets;
		disr_skip = tcp_rskip;
		disr_commit = tcp_rcommit;
		disw_commit = tcp_wcommit;
	}
}

/**
 * @breif
 * 	-DIS_tcp_setup - setup supports routines for dis, "data is strings", to
 * 	use tcp stream I/O.  Also initializes an array of pointers to
 *	buffers and a buffer to be used for the given fd.
 *
 * @param[in] fd - file descriptor
 *
 */

void
DIS_tcp_setup(int fd)
{
#ifdef WIN32
	struct	tcp_chan	*tcp;
	static  int	initialized = 0; /* protected by the tcp lock */
	int			rc;

	/* check for bad file descriptor */
	if (fd < 0)
		return;

	rc = pbs_client_thread_lock_tcp();
	assert(rc == 0);

	/* set DIS function pointers to tcp routines */
	DIS_tcp_funcs();

	if (initialized == 0) {
		DIS_tcparray_init();
		initialized = 1;
	}

	/* allow to search channel without lock - performance */
	/* after few iterations, all channels are created, so
	 * avoiding the lock helps performance */
	if ((tcp=DIS_find_tcp_chan(fd)) == NULL) {

		/* check again to avoid race - if null */
		if ((tcp=DIS_find_tcp_chan(fd)) == NULL)
			tcp = DIS_create_tcp_chan(fd);
	}

	/* initialize read and write buffers */
	DIS_tcp_clear(&tcp->readbuf);
	DIS_tcp_clear(&tcp->writebuf);

	rc = pbs_client_thread_unlock_tcp();
	assert(rc == 0);
#else
	struct	tcp_chan	*tcp;
	struct  tcp_chan	**tmpa;
	int	rc;

	/* check for bad file descriptor */
	if (fd < 0)
		return;

	rc = pbs_client_thread_lock_tcp();
	assert(rc == 0);

	/* set DIS function pointers */
	DIS_tcp_funcs();

	if (fd >= tcparraymax) {
		int	hold = tcparraymax;
		tcparraymax = fd+10;
		if (tcparray == NULL) {
			tcparray = (struct tcp_chan **)
				calloc(tcparraymax,
				sizeof(struct tcp_chan *));
			assert(tcparray != NULL);
		}
		else {
			tmpa = (struct tcp_chan **)realloc(tcparray,
				tcparraymax *
				sizeof(struct tcp_chan *));
			assert(tmpa != NULL);
			tcparray = tmpa;
			memset(&tcparray[hold], '\0',
				(tcparraymax-hold) *
				sizeof(struct tcp_chan *));
		}
	}
	tcp = tcparray[fd];
	if (tcp == NULL) {
		tcp = tcparray[fd] =
			(struct tcp_chan *)malloc(sizeof(struct tcp_chan));
		assert(tcp != NULL);
		tcp->readbuf.tdis_thebuf = malloc(THE_BUF_SIZE);
		assert(tcp->readbuf.tdis_thebuf != NULL);
		tcp->readbuf.tdis_bufsize = THE_BUF_SIZE;
		tcp->writebuf.tdis_thebuf = malloc(THE_BUF_SIZE);
		assert(tcp->writebuf.tdis_thebuf != NULL);
		tcp->writebuf.tdis_bufsize = THE_BUF_SIZE;
	}

	/* initialize read and write buffers */
	DIS_tcp_clear(&tcp->readbuf);
	DIS_tcp_clear(&tcp->writebuf);

	rc = pbs_client_thread_unlock_tcp();
	assert(rc == 0);
#endif
}
