rep_net.c

嵌入式数据库Berkeley DB-4.5.20源代码

	socket_t ns;
	u_int16_t port;

	COMPQUIET(progname, NULL);

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);

	/*
	 * Sites send their listening port when connections are first
	 * established, as it will be different from the outgoing port
	 * for this connection.
	 */
	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 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;
}

/*
 * get_connected_socket --
 *	Connect to the specified port of the specified remote machine,
 *	and return a file descriptor when we have accepted a connection on it.
 *	Add this connection to the machtab.  If we already have a connection
 *	open to this machine, then don't create another one, return the eid
 *	of the connection (in *eidp) and set is_open to 1.  Return 0.
 */
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);
	}

	/*
	 * The first thing we send on the socket is our (listening) port
	 * so the site we are connecting to can register us correctly in
	 * its machtab.
	 */
	nport = htons(myport);
	writesocket(s, &nport, 2);

	return (s);
}

/*
 * get_next_message --
 *	Read a single message from the specified file descriptor, and
 * return it in the format used by rep functions (two DBTs and a type).
 *
 * This function is called in a loop by both clients and masters, and
 * the resulting DBTs are manually dispatched to DB_ENV->rep_process_message().
 */
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;

	/*
	 * The protocol we use on the wire is dead simple:
	 *
	 *	4 bytes		- rec->size
	 *	(# read above)	- rec->data
	 *	4 bytes		- control->size
	 *	(# read above)	- control->data
	 */

	/* Read rec->size. */
	nr = readn(fd, &rsize, 4);
	if (nr != 4)
		return (1);

	/* Read the record itself. */
	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;

	/* Read control->size. */
	nr = readn(fd, &csize, 4);
	if (nr != 4)
		return (1);

	/* Read the control struct itself. */
	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);
}

/*
 * readn --
 *     Read a full n characters from a file descriptor, unless we get an error
 * or EOF.
 */
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) {
			/*
			 * Call read() again on interrupted system call;
			 * on other errors, bail.
			 */
			if (net_errno == EINTR)
				nread = 0;
			else {
				perror("can't read from socket");
				return (-1);
			}
		} else if (nread == 0)
			break;  /* EOF */

		nleft -= nread;
		ptr   += nread;
	}

	return (n - nleft);
}

/*
 * quote_send --
 * The f_send function for DB_ENV->set_rep_transport.
 */
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) {
		/*
		 * Right now, we do not require successful transmission.
		 * I'd like to move this requiring at least one successful
		 * transmission on PERMANENT requests.
		 */
		n = quote_send_broadcast(machtab, rec, control, flags);
		if (n < 0 /*|| (n == 0 && LF_ISSET(DB_REP_PERMANENT))*/)
			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);
}

/*
 * quote_send_broadcast --
 *	Send a message to everybody.
 * Returns the number of sites to which this message was successfully
 * communicated.  A -1 indicates a fatal error.
 */
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);
}

/*
 * quote_send_one --
 *	Send a message to a single machine, given that machine's file
 * descriptor.
 *
 * !!!
 * Note that the machtab mutex should be held through this call.
 * It doubles as a synchronizer to make sure that two threads don't
 * intersperse writes that are part of two single messages.
 */
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);

	/*
	 * The protocol is simply: write rec->size, write rec->data,
	 * write control->size, write control->data.
	 */
	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) {
			/* Try a couple of times to finish the write. */
			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);
}