#include <stdio.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <thread.h>
#include <synch.h>
#include <signal.h>
#include <assert.h>
#include <poll.h>
#include <syslog.h>
#include <libintl.h>
#include "vntsd.h"
#include "chars.h"
typedef struct write_buf {
uint_t sz;
char *buf;
} write_buf_t;
static void
write_chk_status(vntsd_cons_t *consp, int status)
{
if ((consp->status & VNTSD_CONS_DELETED) || (consp->clientpq == NULL)) {
thr_exit(0);
}
switch (status) {
case VNTSD_STATUS_VCC_IO_ERR:
assert(consp->group != NULL);
if (vntsd_vcc_err(consp) != VNTSD_STATUS_CONTINUE) {
thr_exit(0);
}
break;
case VNTSD_STATUS_INTR:
thr_exit(0);
default:
break;
}
}
static int
skip_terminal_null(char *buf, int sz)
{
int i, j;
static int term_null_seq = 0;
assert(sz >= 0);
if (term_null_seq) {
term_null_seq = 0;
if (buf[0] != 0x40) {
for (i = sz; i > 0; i--) {
buf[i] = buf[i-1];
}
buf[0] = 0x5e;
sz++;
} else {
sz--;
if (sz == 0) {
return (sz);
}
for (i = 0; i < sz; i++) {
buf[i] = buf[i+1];
}
}
}
for (; ; ) {
for (i = 0; i < sz; i++) {
if (buf[i] == '\0') {
return (i);
}
if (buf[i] == 0x5e) {
if (i == sz -1) {
term_null_seq = 1;
sz--;
buf[i] = 0;
return (sz);
}
if (buf[i+1] == 0x40) {
sz -= 2;
for (j = i; j < sz -i; j++) {
buf[j] = buf[j+2];
}
break;
}
if (buf[i+1] == '\0') {
buf[i] = 0;
term_null_seq = 1;
return (i);
}
}
}
if (i == sz) {
return (sz);
}
}
}
static int
read_vcc(vntsd_cons_t *consp, char *buf, ssize_t *sz)
{
*sz = read(consp->vcc_fd, buf, VNTSD_MAX_BUF_SIZE);
if (errno == EINTR) {
return (VNTSD_STATUS_INTR);
}
if ((*sz > 0)) {
return (VNTSD_SUCCESS);
}
return (VNTSD_STATUS_VCC_IO_ERR);
}
static boolean_t
write_one_client(vntsd_client_t *clientp, write_buf_t *write_buf)
{
int rv;
rv = vntsd_write_client(clientp, write_buf->buf, write_buf->sz);
if (rv != VNTSD_SUCCESS) {
(void) mutex_lock(&clientp->lock);
clientp->status |= VNTSD_CLIENT_IO_ERR;
assert(clientp->cons);
(void) thr_kill(clientp->cons_tid, 0);
(void) mutex_unlock(&clientp->lock);
}
return (B_FALSE);
}
void*
vntsd_write_thread(vntsd_cons_t *consp)
{
char buf[VNTSD_MAX_BUF_SIZE+1];
int sz;
int rv;
write_buf_t write_buf;
D1(stderr, "t@%d vntsd_write@%d\n", thr_self(), consp->vcc_fd);
assert(consp);
write_chk_status(consp, VNTSD_SUCCESS);
for (; ; ) {
bzero(buf, VNTSD_MAX_BUF_SIZE +1);
rv = read_vcc(consp, buf, &sz);
write_chk_status(consp, rv);
if (sz <= 0) {
continue;
}
if ((sz = skip_terminal_null(buf, sz)) == 0) {
continue;
}
write_buf.sz = sz;
write_buf.buf = buf;
(void) mutex_lock(&consp->lock);
(void) vntsd_que_find(consp->clientpq,
(compare_func_t)write_one_client, &write_buf);
(void) mutex_unlock(&consp->lock);
write_chk_status(consp, VNTSD_SUCCESS);
}
return (NULL);
}