#include <sys/types.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <db.h>
#include "rep_base.h"
#ifndef _SYS_QUEUE_H
#include <dbinc/queue.h>
#endif
int machtab_add __P((machtab_t *, socket_t, u_int32_t, int, int *));
#ifdef DIAGNOSTIC
void machtab_print __P((machtab_t *));
#endif
ssize_t readn __P((socket_t, void *, size_t));
#define MACHID_INVALID 0
#define MACHID_SELF 1
struct __machtab {
LIST_HEAD(__machlist, __member) machlist;
int nextid;
mutex_t mtmutex;
u_int32_t timeout_time;
int current;
int max;
int nsites;
};
struct __member {
u_int32_t hostaddr;
int port;
int eid;
socket_t fd;
LIST_ENTRY(__member) links;
};
static int quote_send_broadcast __P((machtab_t *,
const DBT *, const DBT *, u_int32_t));
static int quote_send_one __P((const DBT *, const DBT *, socket_t, u_int32_t));
int
machtab_init(machtabp, nsites)
machtab_t **machtabp;
int nsites;
{
int ret;
machtab_t *machtab;
if ((machtab = malloc(sizeof(machtab_t))) == NULL) {
fprintf(stderr, "can't allocate memory\n");
return (ENOMEM);
}
LIST_INIT(&machtab->machlist);
machtab->nextid = 2;
machtab->timeout_time = 2 * 1000000;
machtab->current = machtab->max = 0;
machtab->nsites = nsites;
ret = mutex_init(&machtab->mtmutex, NULL);
*machtabp = machtab;
return (ret);
}
int
machtab_add(machtab, fd, hostaddr, port, idp)
machtab_t *machtab;
socket_t fd;
u_int32_t hostaddr;
int port, *idp;
{
int ret;
member_t *m, *member;
ret = 0;
if ((member = malloc(sizeof(member_t))) == NULL) {
fprintf(stderr, "can't allocate memory\n");
return (ENOMEM);
}
member->fd = fd;
member->hostaddr = hostaddr;
member->port = port;
if ((ret = mutex_lock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't lock mutex");
return (ret);
}
for (m = LIST_FIRST(&machtab->machlist);
m != NULL; m = LIST_NEXT(m, links))
if (m->hostaddr == hostaddr && m->port == port)
break;
if (m == NULL) {
member->eid = machtab->nextid++;
LIST_INSERT_HEAD(&machtab->machlist, member, links);
} else
member->eid = m->eid;
if ((ret = mutex_unlock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't unlock mutex\n");
return (ret);
}
if (idp != NULL)
*idp = member->eid;
if (m == NULL) {
if (++machtab->current > machtab->max)
machtab->max = machtab->current;
} else {
free(member);
ret = EEXIST;
}
#ifdef DIAGNOSTIC
printf("Exiting machtab_add\n");
machtab_print(machtab);
#endif
return (ret);
}
int
machtab_getinfo(machtab, eid, hostp, portp)
machtab_t *machtab;
int eid;
u_int32_t *hostp;
int *portp;
{
int ret;
member_t *member;
if ((ret = mutex_lock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't lock mutex\n");
return (ret);
}
for (member = LIST_FIRST(&machtab->machlist);
member != NULL;
member = LIST_NEXT(member, links))
if (member->eid == eid) {
*hostp = member->hostaddr;
*portp = member->port;
break;
}
if ((ret = mutex_unlock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't unlock mutex\n");
return (ret);
}
return (member != NULL ? 0 : EINVAL);
}
int
machtab_rem(machtab, eid, lock)
machtab_t *machtab;
int eid;
int lock;
{
int found, ret;
member_t *member;
ret = 0;
if (lock && (ret = mutex_lock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't lock mutex\n");
return (ret);
}
for (found = 0, member = LIST_FIRST(&machtab->machlist);
member != NULL;
member = LIST_NEXT(member, links))
if (member->eid == eid) {
found = 1;
LIST_REMOVE(member, links);
(void)closesocket(member->fd);
free(member);
machtab->current--;
break;
}
if (LIST_FIRST(&machtab->machlist) == NULL)
machtab->nextid = 2;
if (lock && (ret = mutex_unlock(&machtab->mtmutex)) != 0)
fprintf(stderr, "can't unlock mutex\n");
#ifdef DIAGNOSTIC
printf("Exiting machtab_rem\n");
machtab_print(machtab);
#endif
return (ret);
}
void
machtab_parm(machtab, nump, timeoutp)
machtab_t *machtab;
int *nump;
u_int32_t *timeoutp;
{
if (machtab->nsites == 0)
*nump = machtab->max;
else
*nump = machtab->nsites;
*timeoutp = machtab->timeout_time;
}
#ifdef DIAGNOSTIC
void
machtab_print(machtab)
machtab_t *machtab;
{
member_t *m;
if (mutex_lock(&machtab->mtmutex) != 0) {
fprintf(stderr, "can't lock mutex\n");
abort();
}
for (m = LIST_FIRST(&machtab->machlist);
m != NULL; m = LIST_NEXT(m, links)) {
printf("IP: %lx Port: %6d EID: %2d FD: %3d\n",
(long)m->hostaddr, m->port, m->eid, m->fd);
}
if (mutex_unlock(&machtab->mtmutex) != 0) {
fprintf(stderr, "can't unlock mutex\n");
abort();
}
}
#endif
socket_t
listen_socket_init(progname, port)
const char *progname;
int port;
{
socket_t s;
int sockopt;
struct sockaddr_in si;
COMPQUIET(progname, NULL);
if ((s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
perror("can't create listen socket");
return (-1);
}
memset(&si, 0, sizeof(si));
si.sin_family = AF_INET;
si.sin_addr.s_addr = htonl(INADDR_ANY);
si.sin_port = htons((unsigned short)port);
sockopt = 1;
setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
(const char *)&sockopt, sizeof(sockopt));
if (bind(s, (struct sockaddr *)&si, sizeof(si)) != 0) {
perror("can't bind listen socket");
goto err;
}
if (listen(s, 5) != 0) {
perror("can't establish listen queue");
goto err;
}
return (s);
err: closesocket(s);
return (-1);
}
socket_t
listen_socket_accept(machtab, progname, s, eidp)
machtab_t *machtab;
const char *progname;
socket_t s;
int *eidp;
{
struct sockaddr_in si;
socklen_t si_len;
int host, ret;
socket_t ns;
u_int16_t port;
COMPQUIET(progname, NULL);
accept_wait:
memset(&si, 0, sizeof(si));
si_len = sizeof(si);
ns = accept(s, (struct sockaddr *)&si, &si_len);
if (ns == SOCKET_CREATION_FAILURE) {
fprintf(stderr, "can't accept incoming connection\n");
return ns;
}
host = ntohl(si.sin_addr.s_addr);
if (readn(ns, &port, 2) != 2)
goto err;
port = ntohs(port);
ret = machtab_add(machtab, ns, host, port, eidp);
if (ret == EEXIST) {
closesocket(ns);
goto accept_wait;
} else if (ret != 0)
goto err;
printf("Connected to host %x port %d, eid = %d\n", host, port, *eidp);
return (ns);
err: closesocket(ns);
return SOCKET_CREATION_FAILURE;
}
socket_t
get_connected_socket(machtab, progname, remotehost, port, is_open, eidp)
machtab_t *machtab;
const char *progname, *remotehost;
int port, *is_open, *eidp;
{
int ret;
socket_t s;
struct hostent *hp;
struct sockaddr_in si;
u_int32_t addr;
u_int16_t nport;
*is_open = 0;
if ((hp = gethostbyname(remotehost)) == NULL) {
fprintf(stderr, "%s: host not found: %s\n", progname,
strerror(net_errno));
return (-1);
}
if ((s = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
perror("can't create outgoing socket");
return (-1);
}
memset(&si, 0, sizeof(si));
memcpy((char *)&si.sin_addr, hp->h_addr, hp->h_length);
addr = ntohl(si.sin_addr.s_addr);
ret = machtab_add(machtab, s, addr, port, eidp);
if (ret == EEXIST) {
*is_open = 1;
closesocket(s);
return (0);
} else if (ret != 0) {
closesocket(s);
return (-1);
}
si.sin_family = AF_INET;
si.sin_port = htons((unsigned short)port);
if (connect(s, (struct sockaddr *)&si, sizeof(si)) < 0) {
fprintf(stderr, "%s: connection failed: %s\n",
progname, strerror(net_errno));
(void)machtab_rem(machtab, *eidp, 1);
return (-1);
}
nport = htons(myport);
writesocket(s, &nport, 2);
return (s);
}
int
get_next_message(fd, rec, control)
socket_t fd;
DBT *rec, *control;
{
size_t nr;
u_int32_t rsize, csize;
u_int8_t *recbuf, *controlbuf;
nr = readn(fd, &rsize, 4);
if (nr != 4)
return (1);
if (rsize > 0) {
if (rec->size < rsize)
rec->data = realloc(rec->data, rsize);
recbuf = rec->data;
nr = readn(fd, recbuf, rsize);
} else {
if (rec->data != NULL)
free(rec->data);
rec->data = NULL;
}
rec->size = rsize;
nr = readn(fd, &csize, 4);
if (nr != 4)
return (1);
if (csize > 0) {
controlbuf = control->data;
if (control->size < csize)
controlbuf = realloc(controlbuf, csize);
nr = readn(fd, controlbuf, csize);
if (nr != csize)
return (1);
} else {
if (control->data != NULL)
free(control->data);
controlbuf = NULL;
}
control->data = controlbuf;
control->size = csize;
return (0);
}
ssize_t
readn(fd, vptr, n)
socket_t fd;
void *vptr;
size_t n;
{
size_t nleft;
ssize_t nread;
char *ptr;
ptr = vptr;
nleft = n;
while (nleft > 0) {
if ((nread = readsocket(fd, ptr, nleft)) < 0) {
if (net_errno == EINTR)
nread = 0;
else {
perror("can't read from socket");
return (-1);
}
} else if (nread == 0)
break;
nleft -= nread;
ptr += nread;
}
return (n - nleft);
}
int
quote_send(dbenv, control, rec, lsnp, eid, flags)
DB_ENV *dbenv;
const DBT *control, *rec;
const DB_LSN *lsnp;
int eid;
u_int32_t flags;
{
int n, ret, t_ret;
socket_t fd;
machtab_t *machtab;
member_t *m;
COMPQUIET(lsnp, NULL);
machtab =
(machtab_t *)((APP_DATA*)dbenv->app_private)->comm_infrastructure;
if (eid == DB_EID_BROADCAST) {
n = quote_send_broadcast(machtab, rec, control, flags);
if (n < 0 )
return (DB_REP_UNAVAIL);
return (0);
}
if ((ret = mutex_lock(&machtab->mtmutex)) != 0) {
dbenv->errx(dbenv, "can't lock mutex");
return (ret);
}
fd = 0;
for (m = LIST_FIRST(&machtab->machlist); m != NULL;
m = LIST_NEXT(m, links)) {
if (m->eid == eid) {
fd = m->fd;
break;
}
}
if (fd == 0) {
dbenv->err(dbenv, DB_REP_UNAVAIL,
"quote_send: cannot find machine ID %d", eid);
return (DB_REP_UNAVAIL);
}
if ((ret = quote_send_one(rec, control, fd, flags)) != 0)
fprintf(stderr, "socket write error in send() function\n");
if ((t_ret = mutex_unlock(&machtab->mtmutex)) != 0) {
dbenv->errx(dbenv, "can't unlock mutex");
if (ret == 0)
ret = t_ret;
}
return (ret);
}
static int
quote_send_broadcast(machtab, rec, control, flags)
machtab_t *machtab;
const DBT *rec, *control;
u_int32_t flags;
{
int ret, sent;
member_t *m, *next;
if ((ret = mutex_lock(&machtab->mtmutex)) != 0) {
fprintf(stderr, "can't lock mutex\n");
return (ret);
}
sent = 0;
for (m = LIST_FIRST(&machtab->machlist); m != NULL; m = next) {
next = LIST_NEXT(m, links);
if ((ret = quote_send_one(rec, control, m->fd, flags)) != 0) {
fprintf(stderr, "socket write error in broadcast\n");
(void)machtab_rem(machtab, m->eid, 0);
} else
sent++;
}
if (mutex_unlock(&machtab->mtmutex) != 0) {
fprintf(stderr, "can't unlock mutex\n");
return (-1);
}
return (sent);
}
static int
quote_send_one(rec, control, fd, flags)
const DBT *rec, *control;
socket_t fd;
u_int32_t flags;
{
int retry;
ssize_t bytes_left, nw;
u_int8_t *wp;
COMPQUIET(flags, 0);
nw = writesocket(fd, (const char *)&rec->size, 4);
if (nw != 4)
return (DB_REP_UNAVAIL);
if (rec->size > 0) {
nw = writesocket(fd, rec->data, rec->size);
if (nw < 0)
return (DB_REP_UNAVAIL);
if (nw != (ssize_t)rec->size) {
wp = (u_int8_t *)rec->data + nw;
bytes_left = rec->size - nw;
for (retry = 0; bytes_left > 0 && retry < 3; retry++) {
nw = writesocket(fd, wp, bytes_left);
if (nw < 0)
return (DB_REP_UNAVAIL);
bytes_left -= nw;
wp += nw;
}
if (bytes_left > 0)
return (DB_REP_UNAVAIL);
}
}
nw = writesocket(fd, (const char *)&control->size, 4);
if (nw != 4)
return (DB_REP_UNAVAIL);
if (control->size > 0) {
nw = writesocket(fd, control->data, control->size);
if (nw != (ssize_t)control->size)
return (DB_REP_UNAVAIL);
}
return (0);
}