tsocket: split out a smaller tdgram_context abstraction

The idea is to have a tdgram and a tstream abstraction
which only provide tevent_req based io functions.

metze

4 files changed, 1192 insertions, 0 deletions

diff --git a/lib/tsocket/tsocket.c b/lib/tsocket/tsocket.c
index 1a12e691a9..922429a1c1 100644
--- a/lib/tsocket/tsocket.c
+++ b/lib/tsocket/tsocket.c
@@ -229,3 +229,267 @@ int _tsocket_address_create_socket(const struct tsocket_address *addr,
 	return addr->ops->create_socket(addr, type, mem_ctx, sock, location);
 }
 
+struct tdgram_context {
+	const char *location;
+	const struct tdgram_context_ops *ops;
+	void *private_data;
+};
+
+struct tdgram_context *_tdgram_context_create(TALLOC_CTX *mem_ctx,
+					const struct tdgram_context_ops *ops,
+					void *pstate,
+					size_t psize,
+					const char *type,
+					const char *location)
+{
+	struct tdgram_context *dgram;
+	void **ppstate = (void **)pstate;
+	void *state;
+
+	dgram = talloc(mem_ctx, struct tdgram_context);
+	if (dgram == NULL) {
+		return NULL;
+	}
+	dgram->location	= location;
+	dgram->ops	= ops;
+
+	state = talloc_size(dgram, psize);
+	if (state == NULL) {
+		talloc_free(dgram);
+		return NULL;
+	}
+	talloc_set_name_const(state, type);
+
+	dgram->private_data = state;
+
+	*ppstate = state;
+	return dgram;
+}
+
+void *_tdgram_context_data(struct tdgram_context *dgram)
+{
+	return dgram->private_data;
+}
+
+struct tdgram_recvfrom_state {
+	const struct tdgram_context_ops *ops;
+	uint8_t *buf;
+	size_t len;
+	struct tsocket_address *src;
+};
+
+static void tdgram_recvfrom_done(struct tevent_req *subreq);
+
+struct tevent_req *tdgram_recvfrom_send(TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct tdgram_context *dgram)
+{
+	struct tevent_req *req;
+	struct tdgram_recvfrom_state *state;
+	struct tevent_req *subreq;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_recvfrom_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ops = dgram->ops;
+
+	subreq = state->ops->recvfrom_send(state, ev, dgram);
+	if (tevent_req_nomem(subreq, req)) {
+		goto post;
+	}
+	tevent_req_set_callback(subreq, tdgram_recvfrom_done, req);
+
+	return req;
+
+ post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static void tdgram_recvfrom_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(subreq,
+				 struct tevent_req);
+	struct tdgram_recvfrom_state *state = tevent_req_data(req,
+					      struct tdgram_recvfrom_state);
+	ssize_t ret;
+	int sys_errno;
+
+	ret = state->ops->recvfrom_recv(subreq, &sys_errno, state,
+					&state->buf, &state->src);
+	if (ret == -1) {
+		tevent_req_error(req, sys_errno);
+		return;
+	}
+
+	state->len = ret;
+
+	tevent_req_done(req);
+}
+
+ssize_t tdgram_recvfrom_recv(struct tevent_req *req,
+			     int *perrno,
+			     TALLOC_CTX *mem_ctx,
+			     uint8_t **buf,
+			     struct tsocket_address **src)
+{
+	struct tdgram_recvfrom_state *state = tevent_req_data(req,
+					      struct tdgram_recvfrom_state);
+	ssize_t ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+	if (ret == 0) {
+		*buf = talloc_move(mem_ctx, &state->buf);
+		ret = state->len;
+		if (src) {
+			*src = talloc_move(mem_ctx, &state->src);
+		}
+	}
+
+	tevent_req_received(req);
+	return ret;
+}
+
+struct tdgram_sendto_state {
+	const struct tdgram_context_ops *ops;
+	ssize_t ret;
+};
+
+static void tdgram_sendto_done(struct tevent_req *subreq);
+
+struct tevent_req *tdgram_sendto_send(TALLOC_CTX *mem_ctx,
+				      struct tevent_context *ev,
+				      struct tdgram_context *dgram,
+				      const uint8_t *buf, size_t len,
+				      const struct tsocket_address *dst)
+{
+	struct tevent_req *req;
+	struct tdgram_sendto_state *state;
+	struct tevent_req *subreq;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_sendto_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ops = dgram->ops;
+	state->ret = -1;
+
+	subreq = state->ops->sendto_send(state, ev, dgram,
+					 buf, len, dst);
+	if (tevent_req_nomem(subreq, req)) {
+		goto post;
+	}
+	tevent_req_set_callback(subreq, tdgram_sendto_done, req);
+
+	return req;
+
+ post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static void tdgram_sendto_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(subreq,
+				 struct tevent_req);
+	struct tdgram_sendto_state *state = tevent_req_data(req,
+					    struct tdgram_sendto_state);
+	ssize_t ret;
+	int sys_errno;
+
+	ret = state->ops->sendto_recv(subreq, &sys_errno);
+	if (ret == -1) {
+		tevent_req_error(req, sys_errno);
+		return;
+	}
+
+	state->ret = ret;
+
+	tevent_req_done(req);
+}
+
+ssize_t tdgram_sendto_recv(struct tevent_req *req,
+			   int *perrno)
+{
+	struct tdgram_sendto_state *state = tevent_req_data(req,
+					    struct tdgram_sendto_state);
+	ssize_t ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+	if (ret == 0) {
+		ret = state->ret;
+	}
+
+	tevent_req_received(req);
+	return ret;
+}
+
+struct tdgram_disconnect_state {
+	const struct tdgram_context_ops *ops;
+};
+
+static void tdgram_disconnect_done(struct tevent_req *subreq);
+
+struct tevent_req *tdgram_disconnect_send(TALLOC_CTX *mem_ctx,
+					  struct tevent_context *ev,
+					  struct tdgram_context *dgram)
+{
+	struct tevent_req *req;
+	struct tdgram_disconnect_state *state;
+	struct tevent_req *subreq;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_disconnect_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ops = dgram->ops;
+
+	subreq = state->ops->disconnect_send(state, ev, dgram);
+	if (tevent_req_nomem(subreq, req)) {
+		goto post;
+	}
+	tevent_req_set_callback(subreq, tdgram_disconnect_done, req);
+
+	return req;
+
+ post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static void tdgram_disconnect_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(subreq,
+				 struct tevent_req);
+	struct tdgram_disconnect_state *state = tevent_req_data(req,
+						struct tdgram_disconnect_state);
+	int ret;
+	int sys_errno;
+
+	ret = state->ops->disconnect_recv(subreq, &sys_errno);
+	if (ret == -1) {
+		tevent_req_error(req, sys_errno);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+int tdgram_disconnect_recv(struct tevent_req *req,
+			   int *perrno)
+{
+	int ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+
+	tevent_req_received(req);
+	return ret;
+}
+
diff --git a/lib/tsocket/tsocket.h b/lib/tsocket/tsocket.h
index 9bcfb5cb7e..077fd1ef35 100644
--- a/lib/tsocket/tsocket.h
+++ b/lib/tsocket/tsocket.h
@@ -29,6 +29,7 @@
 
 struct tsocket_context;
 struct tsocket_address;
+struct tdgram_context;
 struct iovec;
 
 enum tsocket_type {
@@ -121,6 +122,32 @@ int _tsocket_address_create_socket(const struct tsocket_address *addr,
 				       __location__)
 
 /*
+ * tdgram_context related functions
+ */
+struct tevent_req *tdgram_recvfrom_send(TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct tdgram_context *dgram);
+ssize_t tdgram_recvfrom_recv(struct tevent_req *req,
+			     int *perrno,
+			     TALLOC_CTX *mem_ctx,
+			     uint8_t **buf,
+			     struct tsocket_address **src);
+
+struct tevent_req *tdgram_sendto_send(TALLOC_CTX *mem_ctx,
+				      struct tevent_context *ev,
+				      struct tdgram_context *dgram,
+				      const uint8_t *buf, size_t len,
+				      const struct tsocket_address *dst);
+ssize_t tdgram_sendto_recv(struct tevent_req *req,
+			   int *perrno);
+
+struct tevent_req *tdgram_disconnect_send(TALLOC_CTX *mem_ctx,
+					  struct tevent_context *ev,
+					  struct tdgram_context *dgram);
+int tdgram_disconnect_recv(struct tevent_req *req,
+			   int *perrno);
+
+/*
  * BSD sockets: inet, inet6 and unix
  */
 
@@ -160,6 +187,22 @@ int _tsocket_context_bsd_wrap_existing(TALLOC_CTX *mem_ctx,
 	_tsocket_context_bsd_wrap_existing(mem_ctx, fd, cod, _sock, \
 					   __location__)
 
+int _tdgram_inet_udp_socket(const struct tsocket_address *local,
+			    const struct tsocket_address *remote,
+			    TALLOC_CTX *mem_ctx,
+			    struct tdgram_context **dgram,
+			    const char *location);
+#define tdgram_inet_udp_socket(local, remote, mem_ctx, dgram) \
+	_tdgram_inet_udp_socket(local, remote, mem_ctx, dgram, __location__)
+
+int _tdgram_unix_dgram_socket(const struct tsocket_address *local,
+			      const struct tsocket_address *remote,
+			      TALLOC_CTX *mem_ctx,
+			      struct tdgram_context **dgram,
+			      const char *location);
+#define tdgram_unix_dgram_socket(local, remote, mem_ctx, dgram) \
+	_tdgram_unix_dgram_socket(local, remote, mem_ctx, dgram, __location__)
+
 /*
  * Async helpers
  */
diff --git a/lib/tsocket/tsocket_bsd.c b/lib/tsocket/tsocket_bsd.c
index 8254f5d9d3..6c60ef2ebd 100644
--- a/lib/tsocket/tsocket_bsd.c
+++ b/lib/tsocket/tsocket_bsd.c
@@ -24,9 +24,168 @@
 #include "replace.h"
 #include "system/filesys.h"
 #include "system/network.h"
+#include "system/filesys.h"
 #include "tsocket.h"
 #include "tsocket_internal.h"
 
+static int tsocket_bsd_error_from_errno(int ret,
+					int sys_errno,
+					bool *retry)
+{
+	*retry = false;
+
+	if (ret >= 0) {
+		return 0;
+	}
+
+	if (ret != -1) {
+		return EIO;
+	}
+
+	if (sys_errno == 0) {
+		return EIO;
+	}
+
+	if (sys_errno == EINTR) {
+		*retry = true;
+		return sys_errno;
+	}
+
+	if (sys_errno == EINPROGRESS) {
+		*retry = true;
+		return sys_errno;
+	}
+
+	if (sys_errno == EAGAIN) {
+		*retry = true;
+		return sys_errno;
+	}
+
+#ifdef EWOULDBLOCK
+	if (sys_errno == EWOULDBLOCK) {
+		*retry = true;
+		return sys_errno;
+	}
+#endif
+
+	return sys_errno;
+}
+
+static int tsocket_bsd_common_prepare_fd(int fd, bool high_fd)
+{
+	int i;
+	int sys_errno = 0;
+	int fds[3];
+	int num_fds = 0;
+
+	int result, flags;
+
+	if (fd == -1) {
+		return -1;
+	}
+
+	/* first make a fd >= 3 */
+	if (high_fd) {
+		while (fd < 3) {
+			fds[num_fds++] = fd;
+			fd = dup(fd);
+			if (fd == -1) {
+				sys_errno = errno;
+				break;
+			}
+		}
+		for (i=0; i<num_fds; i++) {
+			close(fds[i]);
+		}
+		if (fd == -1) {
+			errno = sys_errno;
+			return fd;
+		}
+	}
+
+	/* fd should be nonblocking. */
+
+#ifdef O_NONBLOCK
+#define FLAG_TO_SET O_NONBLOCK
+#else
+#ifdef SYSV
+#define FLAG_TO_SET O_NDELAY
+#else /* BSD */
+#define FLAG_TO_SET FNDELAY
+#endif
+#endif
+
+	if ((flags = fcntl(fd, F_GETFL)) == -1) {
+		goto fail;
+	}
+
+	flags |= FLAG_TO_SET;
+	if (fcntl(fd, F_SETFL, flags) == -1) {
+		goto fail;
+	}
+
+#undef FLAG_TO_SET
+
+	/* fd should be closed on exec() */
+#ifdef FD_CLOEXEC
+	result = flags = fcntl(fd, F_GETFD, 0);
+	if (flags >= 0) {
+		flags |= FD_CLOEXEC;
+		result = fcntl(fd, F_SETFD, flags);
+	}
+	if (result < 0) {
+		goto fail;
+	}
+#endif
+	return fd;
+
+ fail:
+	if (fd != -1) {
+		sys_errno = errno;
+		close(fd);
+		errno = sys_errno;
+	}
+	return -1;
+}
+
+static ssize_t tsocket_bsd_pending(int fd)
+{
+	int ret;
+	int value = 0;
+
+	ret = ioctl(fd, FIONREAD, &value);
+	if (ret == -1) {
+		return ret;
+	}
+
+	if (ret == 0) {
+		if (value == 0) {
+			int error=0;
+			socklen_t len = sizeof(error);
+			/*
+			 * if no data is available check if the socket
+			 * is in error state. For dgram sockets
+			 * it's the way to return ICMP error messages
+			 * of connected sockets to the caller.
+			 */
+			ret = getsockopt(fd, SOL_SOCKET, SO_ERROR,
+					 &error, &len);
+			if (ret == -1) {
+				return ret;
+			}
+			if (error != 0) {
+				errno = error;
+				return -1;
+			}
+		}
+		return value;
+	}
+
+	/* this should not be reached */
+	errno = EIO;
+	return -1;
+}
+
 static const struct tsocket_context_ops tsocket_context_bsd_ops;
 static const struct tsocket_address_ops tsocket_address_bsd_ops;
 
@@ -1125,3 +1284,688 @@ static const struct tsocket_context_ops tsocket_context_bsd_ops = {
 
 	.disconnect		= tsocket_context_bsd_disconnect
 };
+
+struct tdgram_bsd {
+	int fd;
+
+	void *event_ptr;
+	struct tevent_fd *fde;
+
+	void *readable_private;
+	void (*readable_handler)(void *private_data);
+	void *writeable_private;
+	void (*writeable_handler)(void *private_data);
+
+	struct tevent_req *read_req;
+	struct tevent_req *write_req;
+};
+
+static void tdgram_bsd_fde_handler(struct tevent_context *ev,
+				   struct tevent_fd *fde,
+				   uint16_t flags,
+				   void *private_data)
+{
+	struct tdgram_bsd *bsds = talloc_get_type_abort(private_data,
+				  struct tdgram_bsd);
+
+	if (flags & TEVENT_FD_WRITE) {
+		bsds->writeable_handler(bsds->writeable_private);
+		return;
+	}
+	if (flags & TEVENT_FD_READ) {
+		bsds->readable_handler(bsds->readable_private);
+		return;
+	}
+}
+
+static int tdgram_bsd_set_readable_handler(struct tdgram_bsd *bsds,
+					   struct tevent_context *ev,
+					   void (*handler)(void *private_data),
+					   void *private_data)
+{
+	if (ev == NULL) {
+		if (handler) {
+			errno = EINVAL;
+			return -1;
+		}
+
+		bsds->readable_handler = NULL;
+		bsds->readable_private = NULL;
+		TEVENT_FD_NOT_READABLE(bsds->fde);
+
+		if (bsds->fde && !bsds->writeable_handler) {
+			/* we don't need the fd event anymore */
+			bsds->event_ptr = NULL;
+			TALLOC_FREE(bsds->fde);
+		}
+		return 0;
+	}
+
+	if (bsds->fde == NULL) {
+		bsds->fde = tevent_add_fd(ev, bsds,
+					  bsds->fd, TEVENT_FD_READ,
+					  tdgram_bsd_fde_handler,
+					  bsds);
+		if (!bsds->fde) {
+			return -1;
+		}
+
+		/* cache the event context we're running on */
+		bsds->event_ptr = ev;
+	}
+
+	/* read and write must use the same tevent_context */
+	if (bsds->event_ptr != ev) {
+		errno = EINVAL;
+		return -1;
+	}
+
+	TEVENT_FD_READABLE(bsds->fde);
+	bsds->readable_handler = handler;
+	bsds->readable_private = private_data;
+
+	return 0;
+}
+
+static int tdgram_bsd_set_writeable_handler(struct tdgram_bsd *bsds,
+					    struct tevent_context *ev,
+					    void (*handler)(void *private_data),
+					    void *private_data)
+{
+	if (ev == NULL) {
+		if (handler) {
+			errno = EINVAL;
+			return -1;
+		}
+
+		bsds->writeable_handler = NULL;
+		bsds->writeable_private = NULL;
+		TEVENT_FD_NOT_WRITEABLE(bsds->fde);
+
+		if (bsds->fde && !bsds->readable_handler) {
+			/* we don't need the fd event anymore */
+			bsds->event_ptr = NULL;
+			TALLOC_FREE(bsds->fde);
+		}
+		return 0;
+	}
+
+	if (bsds->fde == NULL) {
+		bsds->fde = tevent_add_fd(ev, bsds,
+					  bsds->fd, TEVENT_FD_WRITE,
+					  tdgram_bsd_fde_handler,
+					  bsds);
+		if (!bsds->fde) {
+			return -1;
+		}
+
+		/* cache the event context we're running on */
+		bsds->event_ptr = ev;
+	}
+
+	/* read and write must use the same tevent_context */
+	if (bsds->event_ptr != ev) {
+		errno = EINVAL;
+		return -1;
+	}
+
+	TEVENT_FD_WRITEABLE(bsds->fde);
+	bsds->writeable_handler = handler;
+	bsds->writeable_private = private_data;
+
+	return 0;
+}
+
+struct tdgram_bsd_recvfrom_state {
+	struct tdgram_context *dgram;
+
+	uint8_t *buf;
+	size_t len;
+	struct tsocket_address *src;
+};
+
+static int tdgram_bsd_recvfrom_destructor(struct tdgram_bsd_recvfrom_state *state)
+{
+	struct tdgram_bsd *bsds = tdgram_context_data(state->dgram,
+				  struct tdgram_bsd);
+
+	bsds->read_req = NULL;
+	tdgram_bsd_set_readable_handler(bsds, NULL, NULL, NULL);
+
+	return 0;
+}
+
+static void tdgram_bsd_recvfrom_handler(void *private_data);
+
+static struct tevent_req *tdgram_bsd_recvfrom_send(TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct tdgram_context *dgram)
+{
+	struct tevent_req *req;
+	struct tdgram_bsd_recvfrom_state *state;
+	struct tdgram_bsd *bsds = tdgram_context_data(dgram, struct tdgram_bsd);
+	int ret;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_bsd_recvfrom_state);
+	if (!req) {
+		return NULL;
+	}
+
+	state->dgram	= dgram;
+	state->buf	= NULL;
+	state->len	= 0;
+	state->src	= NULL;
+
+	if (bsds->read_req) {
+		tevent_req_error(req, EBUSY);
+		goto post;
+	}
+	bsds->read_req = req;
+
+	talloc_set_destructor(state, tdgram_bsd_recvfrom_destructor);
+
+	if (bsds->fd == -1) {
+		tevent_req_error(req, ENOTCONN);
+		goto post;
+	}
+
+	ret = tdgram_bsd_set_readable_handler(bsds, ev,
+					      tdgram_bsd_recvfrom_handler,
+					      req);
+	if (ret == -1) {
+		tevent_req_error(req, errno);
+		goto post;
+	}
+
+	return req;
+
+ post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static void tdgram_bsd_recvfrom_handler(void *private_data)
+{
+	struct tevent_req *req = talloc_get_type_abort(private_data,
+				 struct tevent_req);
+	struct tdgram_bsd_recvfrom_state *state = tevent_req_data(req,
+					struct tdgram_bsd_recvfrom_state);
+	struct tdgram_context *dgram = state->dgram;
+	struct tdgram_bsd *bsds = tdgram_context_data(dgram, struct tdgram_bsd);
+	struct tsocket_address_bsd *bsda;
+	ssize_t ret;
+	struct sockaddr *sa = NULL;
+	socklen_t sa_len = 0;
+	int err;
+	bool retry;
+
+	ret = tsocket_bsd_pending(bsds->fd);
+	if (ret == 0) {
+		/* retry later */
+		return;
+	}
+	err = tsocket_bsd_error_from_errno(ret, errno, &retry);
+	if (retry) {
+		/* retry later */
+		return;
+	}
+	if (tevent_req_error(req, err)) {
+		return;
+	}
+
+	state->buf = talloc_array(state, uint8_t, ret);
+	if (tevent_req_nomem(state->buf, req)) {
+		return;
+	}
+	state->len = ret;
+
+	state->src = tsocket_address_create(state,
+					    &tsocket_address_bsd_ops,
+					    &bsda,
+					    struct tsocket_address_bsd,
+					    __location__ "bsd_recvfrom");
+	if (tevent_req_nomem(state->src, req)) {
+		return;
+	}
+
+	ZERO_STRUCTP(bsda);
+
+	sa = &bsda->u.sa;
+	sa_len = sizeof(bsda->u.ss);
+
+	ret = recvfrom(bsds->fd, state->buf, state->len, 0, sa, &sa_len);
+	err = tsocket_error_from_errno(ret, errno, &retry);
+	if (retry) {
+		/* retry later */
+		return;
+	}
+	if (tevent_req_error(req, err)) {
+		return;
+	}
+
+	if (ret != state->len) {
+		tevent_req_error(req, EIO);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static ssize_t tdgram_bsd_recvfrom_recv(struct tevent_req *req,
+					int *perrno,
+					TALLOC_CTX *mem_ctx,
+					uint8_t **buf,
+					struct tsocket_address **src)
+{
+	struct tdgram_bsd_recvfrom_state *state = tevent_req_data(req,
+					struct tdgram_bsd_recvfrom_state);
+	ssize_t ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+	if (ret == 0) {
+		*buf = talloc_move(mem_ctx, &state->buf);
+		ret = state->len;
+		if (src) {
+			*src = talloc_move(mem_ctx, &state->src);
+		}
+	}
+
+	tevent_req_received(req);
+	return ret;
+}
+
+struct tdgram_bsd_sendto_state {
+	struct tdgram_context *dgram;
+
+	const uint8_t *buf;
+	size_t len;
+	const struct tsocket_address *dst;
+
+	ssize_t ret;
+};
+
+static int tdgram_bsd_sendto_destructor(struct tdgram_bsd_sendto_state *state)
+{
+	struct tdgram_bsd *bsds = tdgram_context_data(state->dgram,
+				  struct tdgram_bsd);
+
+	bsds->write_req = NULL;
+	tdgram_bsd_set_writeable_handler(bsds, NULL, NULL, NULL);
+	return 0;
+}
+
+static void tdgram_bsd_sendto_handler(void *private_data);
+
+static struct tevent_req *tdgram_bsd_sendto_send(TALLOC_CTX *mem_ctx,
+						 struct tevent_context *ev,
+						 struct tdgram_context *dgram,
+						 const uint8_t *buf,
+						 size_t len,
+						 const struct tsocket_address *dst)
+{
+	struct tevent_req *req;
+	struct tdgram_bsd_sendto_state *state;
+	struct tdgram_bsd *bsds = tdgram_context_data(dgram, struct tdgram_bsd);
+	int ret;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_bsd_sendto_state);
+	if (!req) {
+		return NULL;
+	}
+
+	state->dgram	= dgram;
+	state->buf	= buf;
+	state->len	= len;
+	state->dst	= dst;
+	state->ret	= -1;
+
+	if (bsds->write_req) {
+		tevent_req_error(req, EBUSY);
+		goto post;
+	}
+	bsds->write_req = req;
+
+	talloc_set_destructor(state, tdgram_bsd_sendto_destructor);
+
+	if (bsds->fd == -1) {
+		tevent_req_error(req, ENOTCONN);
+		goto post;
+	}
+
+	ret = tdgram_bsd_set_writeable_handler(bsds, ev,
+					       tdgram_bsd_sendto_handler,
+					       req);
+	if (ret == -1) {
+		tevent_req_error(req, errno);
+		goto post;
+	}
+
+	return req;
+
+ post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static void tdgram_bsd_sendto_handler(void *private_data)
+{
+	struct tevent_req *req = talloc_get_type_abort(private_data,
+				 struct tevent_req);
+	struct tdgram_bsd_sendto_state *state = tevent_req_data(req,
+					struct tdgram_bsd_sendto_state);
+	struct tdgram_context *dgram = state->dgram;
+	struct tdgram_bsd *bsds = tdgram_context_data(dgram, struct tdgram_bsd);
+	struct sockaddr *sa = NULL;
+	socklen_t sa_len = 0;
+	ssize_t ret;
+	int err;
+	bool retry;
+
+	if (state->dst) {
+		struct tsocket_address_bsd *bsda =
+			talloc_get_type(state->dst->private_data,
+			struct tsocket_address_bsd);
+
+		sa = &bsda->u.sa;
+		sa_len = sizeof(bsda->u.ss);
+	}
+
+	ret = sendto(bsds->fd, state->buf, state->len, 0, sa, sa_len);
+	err = tsocket_error_from_errno(ret, errno, &retry);
+	if (retry) {
+		/* retry later */
+		return;
+	}
+	if (tevent_req_error(req, err)) {
+		return;
+	}
+
+	state->ret = ret;
+
+	tevent_req_done(req);
+}
+
+static ssize_t tdgram_bsd_sendto_recv(struct tevent_req *req, int *perrno)
+{
+	struct tdgram_bsd_sendto_state *state = tevent_req_data(req,
+					struct tdgram_bsd_sendto_state);
+	ssize_t ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+	if (ret == 0) {
+		ret = state->ret;
+	}
+
+	tevent_req_received(req);
+	return ret;
+}
+
+struct tdgram_bsd_disconnect_state {
+	int ret;
+};
+
+static struct tevent_req *tdgram_bsd_disconnect_send(TALLOC_CTX *mem_ctx,
+						     struct tevent_context *ev,
+						     struct tdgram_context *dgram)
+{
+	struct tdgram_bsd *bsds = tdgram_context_data(dgram, struct tdgram_bsd);
+	struct tevent_req *req;
+	struct tdgram_bsd_disconnect_state *state;
+	int ret;
+	int err;
+	bool dummy;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct tdgram_bsd_disconnect_state);
+	if (req == NULL) {
+		return NULL;
+	}
+	state->ret = -1;
+
+	if (bsds->read_req || bsds->write_req) {
+		tevent_req_error(req, EBUSY);
+		goto post;
+	}
+
+	if (bsds->fd == -1) {
+		tevent_req_error(req, ENOTCONN);
+		goto post;
+	}
+
+	state->ret = close(bsds->fd);
+	bsds->fd = -1;
+	err = tsocket_error_from_errno(ret, errno, &dummy);
+	if (tevent_req_error(req, err)) {
+		goto post;
+	}
+
+	tevent_req_done(req);
+post:
+	tevent_req_post(req, ev);
+	return req;
+}
+
+static int tdgram_bsd_disconnect_recv(struct tevent_req *req,
+				      int *perrno)
+{
+	struct tdgram_bsd_disconnect_state *state = tevent_req_data(req,
+					struct tdgram_bsd_disconnect_state);
+	int ret;
+
+	ret = tsocket_simple_int_recv(req, perrno);
+	if (ret == 0) {
+		ret = state->ret;
+	}
+
+	tevent_req_received(req);
+	return ret;
+}
+
+static const struct tdgram_context_ops tdgram_bsd_ops = {
+	.name			= "bsd",
+
+	.recvfrom_send		= tdgram_bsd_recvfrom_send,
+	.recvfrom_recv		= tdgram_bsd_recvfrom_recv,
+
+	.sendto_send		= tdgram_bsd_sendto_send,
+	.sendto_recv		= tdgram_bsd_sendto_recv,
+
+	.disconnect_send	= tdgram_bsd_disconnect_send,
+	.disconnect_recv	= tdgram_bsd_disconnect_recv,
+};
+
+static int tdgram_bsd_destructor(struct tdgram_bsd *bsds)
+{
+	TALLOC_FREE(bsds->fde);
+	if (bsds->fd != -1) {
+		close(bsds->fd);
+		bsds->fd = -1;
+	}
+	return 0;
+}
+
+static int tdgram_bsd_dgram_socket(const struct tsocket_address *local,
+				   const struct tsocket_address *remote,
+				   TALLOC_CTX *mem_ctx,
+				   struct tdgram_context **_dgram,
+				   const char *location)
+{
+	struct tsocket_address_bsd *lbsda =
+		talloc_get_type_abort(local->private_data,
+		struct tsocket_address_bsd);
+	struct tsocket_address_bsd *rbsda = NULL;
+	struct tdgram_context *dgram;
+	struct tdgram_bsd *bsds;
+	int fd;
+	int ret;
+	bool do_bind = false;
+	bool do_reuseaddr = false;
+
+	if (remote) {
+		rbsda = talloc_get_type_abort(remote->private_data,
+			struct tsocket_address_bsd);
+	}
+
+	switch (lbsda->u.sa.sa_family) {
+	case AF_UNIX:
+		if (lbsda->u.un.sun_path[0] != 0) {
+			do_reuseaddr = true;
+			do_bind = true;
+		}
+		break;
+	case AF_INET:
+		if (lbsda->u.in.sin_port != 0) {
+			do_reuseaddr = true;
+			do_bind = true;
+		}
+		if (lbsda->u.in.sin_addr.s_addr == INADDR_ANY) {
+			do_bind = true;
+		}
+		break;
+#ifdef HAVE_IPV6
+	case AF_INET6:
+		if (lbsda->u.in6.sin6_port != 0) {
+			do_reuseaddr = true;
+			do_bind = true;
+		}
+		if (memcmp(&in6addr_any,
+			   &lbsda->u.in6.sin6_addr,
+			   sizeof(in6addr_any)) != 0) {
+			do_bind = true;
+		}
+		break;
+#endif
+	default:
+		errno = EINVAL;
+		return -1;
+	}
+
+	fd = socket(lbsda->u.sa.sa_family, SOCK_DGRAM, 0);
+	if (fd < 0) {
+		return fd;
+	}
+
+	fd = tsocket_bsd_common_prepare_fd(fd, true);
+	if (fd < 0) {
+		return fd;
+	}
+
+	dgram = tdgram_context_create(mem_ctx,
+				      &tdgram_bsd_ops,
+				      &bsds,
+				      struct tdgram_bsd,
+				      location);
+	if (!dgram) {
+		int saved_errno = errno;
+		close(fd);
+		errno = saved_errno;
+		return -1;
+	}
+	ZERO_STRUCTP(bsds);
+	bsds->fd = fd;
+	talloc_set_destructor(bsds, tdgram_bsd_destructor);
+
+	if (lbsda->broadcast) {
+		int val = 1;
+
+		ret = setsockopt(fd, SOL_SOCKET, SO_BROADCAST,
+				 (const void *)&val, sizeof(val));
+		if (ret == -1) {
+			int saved_errno = errno;
+			talloc_free(dgram);
+			errno = saved_errno;
+			return ret;
+		}
+	}
+
+	if (do_reuseaddr) {
+		int val = 1;
+
+		ret = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
+				 (const void *)&val, sizeof(val));
+		if (ret == -1) {
+			int saved_errno = errno;
+			talloc_free(dgram);
+			errno = saved_errno;
+			return ret;
+		}
+	}
+
+	if (do_bind) {
+		ret = bind(fd, &lbsda->u.sa, sizeof(lbsda->u.ss));
+		if (ret == -1) {
+			int saved_errno = errno;
+			talloc_free(dgram);
+			errno = saved_errno;
+			return ret;
+		}
+	}
+
+	if (rbsda) {
+		ret = connect(fd, &rbsda->u.sa, sizeof(rbsda->u.ss));
+		if (ret == -1) {
+			int saved_errno = errno;
+			talloc_free(dgram);
+			errno = saved_errno;
+			return ret;
+		}
+	}
+
+	*_dgram = dgram;
+	return 0;
+}
+
+int _tdgram_inet_udp_socket(const struct tsocket_address *local,
+			    const struct tsocket_address *remote,
+			    TALLOC_CTX *mem_ctx,
+			    struct tdgram_context **dgram,
+			    const char *location)
+{
+	struct tsocket_address_bsd *lbsda =
+		talloc_get_type_abort(local->private_data,
+		struct tsocket_address_bsd);
+	int ret;
+
+	switch (lbsda->u.sa.sa_family) {
+	case AF_INET:
+		break;
+#ifdef HAVE_IPV6
+	case AF_INET6:
+		break;
+#endif
+	default:
+		errno = EINVAL;
+		return -1;
+	}
+
+	ret = tdgram_bsd_dgram_socket(local, remote, mem_ctx, dgram, location);
+
+	return ret;
+}
+
+int _tdgram_unix_dgram_socket(const struct tsocket_address *local,
+			      const struct tsocket_address *remote,
+			      TALLOC_CTX *mem_ctx,
+			      struct tdgram_context **dgram,
+			      const char *location)
+{
+	struct tsocket_address_bsd *lbsda =
+		talloc_get_type_abort(local->private_data,
+		struct tsocket_address_bsd);
+	int ret;
+
+	switch (lbsda->u.sa.sa_family) {
+	case AF_UNIX:
+		break;
+	default:
+		errno = EINVAL;
+		return -1;
+	}
+
+	ret = tdgram_bsd_dgram_socket(local, remote, mem_ctx, dgram, location);
+
+	return ret;
+}
+
diff --git a/lib/tsocket/tsocket_internal.h b/lib/tsocket/tsocket_internal.h
index e4a4908f3e..d1f240eba0 100644
--- a/lib/tsocket/tsocket_internal.h
+++ b/lib/tsocket/tsocket_internal.h
@@ -149,6 +149,47 @@ struct tsocket_address *_tsocket_address_create(TALLOC_CTX *mem_ctx,
 	_tsocket_address_create(mem_ctx, ops, state, sizeof(type), \
 				#type, location)
 
+struct tdgram_context_ops {
+	const char *name;
+
+	struct tevent_req *(*recvfrom_send)(TALLOC_CTX *mem_ctx,
+					    struct tevent_context *ev,
+					    struct tdgram_context *dgram);
+	ssize_t (*recvfrom_recv)(struct tevent_req *req,
+				 int *perrno,
+				 TALLOC_CTX *mem_ctx,
+				 uint8_t **buf,
+				 struct tsocket_address **src);
+
+	struct tevent_req *(*sendto_send)(TALLOC_CTX *mem_ctx,
+					  struct tevent_context *ev,
+					  struct tdgram_context *dgram,
+					  const uint8_t *buf, size_t len,
+					  const struct tsocket_address *dst);
+	ssize_t (*sendto_recv)(struct tevent_req *req,
+			       int *perrno);
+
+	struct tevent_req *(*disconnect_send)(TALLOC_CTX *mem_ctx,
+					      struct tevent_context *ev,
+					      struct tdgram_context *dgram);
+	int (*disconnect_recv)(struct tevent_req *req,
+			       int *perrno);
+};
+
+struct tdgram_context *_tdgram_context_create(TALLOC_CTX *mem_ctx,
+					const struct tdgram_context_ops *ops,
+					void *pstate,
+					size_t psize,
+					const char *type,
+					const char *location);
+#define tdgram_context_create(mem_ctx, ops, state, type, location) \
+	_tdgram_context_create(mem_ctx, ops, state, sizeof(type), \
+				#type, location)
+
+void *_tdgram_context_data(struct tdgram_context *dgram);
+#define tdgram_context_data(_req, _type) \
+	talloc_get_type_abort(_tdgram_context_data(_req), _type)
+
 int tsocket_error_from_errno(int ret, int sys_errno, bool *retry);
 int tsocket_simple_int_recv(struct tevent_req *req, int *perrno);
 int tsocket_common_prepare_fd(int fd, bool high_fd);