--- /dev/null
+/*
+ Unix SMB/CIFS implementation.
+ Samba3 ctdb connection handling
+ Copyright (C) Volker Lendecke 2012
+
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ This program 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "lib/util/tevent_unix.h"
+#include "ctdb_conn.h"
+
+#ifdef CLUSTER_SUPPORT
+
+#include "lib/async_req/async_sock.h"
+
+struct ctdb_conn {
+ int fd;
+ struct tevent_queue *outqueue;
+};
+
+struct ctdb_conn_init_state {
+ struct sockaddr_un addr;
+ struct ctdb_conn *conn;
+};
+
+static void ctdb_conn_init_done(struct tevent_req *subreq);
+static int ctdb_conn_destructor(struct ctdb_conn *conn);
+
+struct tevent_req *ctdb_conn_init_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ const char *sock)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_conn_init_state *state;
+
+ req = tevent_req_create(mem_ctx, &state, struct ctdb_conn_init_state);
+ if (req == NULL) {
+ return NULL;
+ }
+
+ if (!lp_clustering()) {
+ tevent_req_error(req, ENOSYS);
+ return tevent_req_post(req, ev);
+ }
+
+ if (strlen(sock) >= sizeof(state->addr.sun_path)) {
+ tevent_req_error(req, ENAMETOOLONG);
+ return tevent_req_post(req, ev);
+ }
+
+ state->conn = talloc(state, struct ctdb_conn);
+ if (tevent_req_nomem(state->conn, req)) {
+ return tevent_req_post(req, ev);
+ }
+
+ state->conn->outqueue = tevent_queue_create(
+ state->conn, "ctdb outqueue");
+ if (tevent_req_nomem(state->conn->outqueue, req)) {
+ return tevent_req_post(req, ev);
+ }
+
+ state->conn->fd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (state->conn->fd == -1) {
+ tevent_req_error(req, errno);
+ return tevent_req_post(req, ev);
+ }
+ talloc_set_destructor(state->conn, ctdb_conn_destructor);
+
+ state->addr.sun_family = AF_UNIX;
+ strncpy(state->addr.sun_path, sock, sizeof(state->addr.sun_path));
+
+ subreq = async_connect_send(state, ev, state->conn->fd,
+ (struct sockaddr *)&state->addr,
+ sizeof(state->addr));
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_init_done, req);
+ return req;
+}
+
+static int ctdb_conn_destructor(struct ctdb_conn *c)
+{
+ if (c->fd != -1) {
+ close(c->fd);
+ c->fd = -1;
+ }
+ return 0;
+}
+
+static void ctdb_conn_init_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ int ret, err;
+
+ ret = async_connect_recv(subreq, &err);
+ TALLOC_FREE(subreq);
+ if (ret == -1) {
+ tevent_req_error(req, err);
+ return;
+ }
+ tevent_req_done(req);
+}
+
+int ctdb_conn_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct ctdb_conn **pconn)
+{
+ struct ctdb_conn_init_state *state = tevent_req_data(
+ req, struct ctdb_conn_init_state);
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ *pconn = talloc_move(mem_ctx, &state->conn);
+
+ return 0;
+}
+
+struct ctdb_conn_control_state {
+ struct tevent_context *ev;
+ struct ctdb_conn *conn;
+ struct ctdb_req_control req;
+ struct iovec iov[2];
+ struct ctdb_reply_control *reply;
+};
+
+static void ctdb_conn_control_written(struct tevent_req *subreq);
+static void ctdb_conn_control_done(struct tevent_req *subreq);
+static ssize_t ctdb_packet_more(uint8_t *buf, size_t buflen, void *p);
+
+struct tevent_req *ctdb_conn_control_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct ctdb_conn *conn,
+ uint32_t vnn, uint32_t opcode,
+ uint64_t srvid, uint32_t flags,
+ uint8_t *data, size_t datalen)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_conn_control_state *state;
+ struct ctdb_req_header *hdr;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct ctdb_conn_control_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->ev = ev;
+ state->conn = conn;
+
+ hdr = &state->req.hdr;
+ hdr->length = offsetof(struct ctdb_req_control, data) + datalen;
+ hdr->ctdb_magic = CTDB_MAGIC;
+ hdr->ctdb_version = CTDB_VERSION;
+ hdr->operation = CTDB_REQ_CONTROL;
+ hdr->reqid = 1; /* FIXME */
+ hdr->destnode = vnn;
+ state->req.opcode = opcode;
+ state->req.srvid = srvid;
+ state->req.datalen = datalen;
+ state->req.flags = flags;
+
+ state->iov[0].iov_base = &state->req;
+ state->iov[0].iov_len = offsetof(struct ctdb_req_control, data);
+ state->iov[1].iov_base = data;
+ state->iov[1].iov_len = datalen;
+
+ subreq = writev_send(state, ev, conn->outqueue, conn->fd, false,
+ state->iov, 2);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_control_written, req);
+ return req;
+}
+
+static void ctdb_conn_control_written(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_control_state *state = tevent_req_data(
+ req, struct ctdb_conn_control_state);
+ ssize_t written;
+ int err;
+
+ written = writev_recv(subreq, &err);
+ TALLOC_FREE(subreq);
+ if (written == -1) {
+ tevent_req_error(req, err);
+ return;
+ }
+ subreq = read_packet_send(
+ state, state->ev, state->conn->fd, sizeof(uint32_t),
+ ctdb_packet_more, NULL);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_control_done, req);
+}
+
+static ssize_t ctdb_packet_more(uint8_t *buf, size_t buflen, void *p)
+{
+ uint32_t len;
+
+ if (buflen > sizeof(uint32_t)) {
+ /* Been here, done */
+ return 0;
+ }
+ memcpy(&len, buf, sizeof(len));
+ return (len - sizeof(uint32_t));
+}
+
+static void ctdb_conn_control_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_control_state *state = tevent_req_data(
+ req, struct ctdb_conn_control_state);
+ ssize_t nread;
+ uint8_t *buf;
+ int err;
+
+ nread = read_packet_recv(subreq, state, &buf, &err);
+ TALLOC_FREE(subreq);
+ if (nread == -1) {
+ tevent_req_error(req, err);
+ return;
+ }
+ state->reply = (struct ctdb_reply_control *)buf;
+ tevent_req_done(req);
+}
+
+int ctdb_conn_control_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct ctdb_reply_control **preply)
+{
+ struct ctdb_conn_control_state *state = tevent_req_data(
+ req, struct ctdb_conn_control_state);
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ if (preply != NULL) {
+ *preply = talloc_move(mem_ctx, &state->reply);
+ }
+ return 0;
+}
+
+struct ctdb_conn_msg_write_state {
+ struct ctdb_req_message ctdb_msg;
+ struct iovec iov[2];
+};
+
+static void ctdb_conn_msg_write_done(struct tevent_req *subreq);
+
+struct tevent_req *ctdb_conn_msg_write_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct ctdb_conn *conn,
+ uint32_t vnn, uint64_t srvid,
+ uint8_t *msg, size_t msg_len)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_conn_msg_write_state *state;
+ struct ctdb_req_header *h;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct ctdb_conn_msg_write_state);
+ if (req == NULL) {
+ return NULL;
+ }
+
+ h = &state->ctdb_msg.hdr;
+
+ h->length = offsetof(struct ctdb_req_message, data) + msg_len;
+ h->ctdb_magic = CTDB_MAGIC;
+ h->ctdb_version = CTDB_VERSION;
+ h->generation = 1;
+ h->operation = CTDB_REQ_MESSAGE;
+ h->destnode = vnn;
+ h->srcnode = CTDB_CURRENT_NODE;
+ h->reqid = 0;
+ state->ctdb_msg.srvid = srvid;
+ state->ctdb_msg.datalen = msg_len;
+
+ state->iov[0].iov_base = &state->ctdb_msg;
+ state->iov[0].iov_len = offsetof(struct ctdb_req_message, data);
+ state->iov[1].iov_base = msg;
+ state->iov[1].iov_len = msg_len;
+
+ subreq = writev_send(state, ev, conn->outqueue, conn->fd, false,
+ state->iov, 2);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_msg_write_done, req);
+ return req;
+}
+
+static void ctdb_conn_msg_write_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ ssize_t written;
+ int err;
+
+ written = writev_recv(subreq, &err);
+ TALLOC_FREE(subreq);
+ if (written == -1) {
+ tevent_req_error(req, err);
+ return;
+ }
+ tevent_req_done(req);
+}
+
+int ctdb_conn_msg_write_recv(struct tevent_req *req)
+{
+ int err;
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ return 0;
+}
+
+struct ctdb_msg_channel {
+ struct ctdb_conn *conn;
+};
+
+struct ctdb_msg_channel_init_state {
+ struct tevent_context *ev;
+ struct ctdb_conn *conn;
+ uint64_t srvid;
+ struct ctdb_msg_channel *channel;
+};
+
+static void ctdb_msg_channel_init_connected(struct tevent_req *subreq);
+static void ctdb_msg_channel_init_registered_srvid(struct tevent_req *subreq);
+
+struct tevent_req *ctdb_msg_channel_init_send(
+ TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+ const char *sock, uint64_t srvid)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_msg_channel_init_state *state;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct ctdb_msg_channel_init_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->ev = ev;
+ state->srvid = srvid;
+
+ subreq = ctdb_conn_init_send(state, ev, sock);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_msg_channel_init_connected, req);
+ return req;
+}
+
+static void ctdb_msg_channel_init_connected(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_msg_channel_init_state *state = tevent_req_data(
+ req, struct ctdb_msg_channel_init_state);
+ int ret;
+
+ ret = ctdb_conn_init_recv(subreq, state, &state->conn);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ subreq = ctdb_conn_control_send(state, state->ev, state->conn,
+ CTDB_CURRENT_NODE,
+ CTDB_CONTROL_REGISTER_SRVID,
+ state->srvid, 0, NULL, 0);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(
+ subreq, ctdb_msg_channel_init_registered_srvid, req);
+}
+
+static void ctdb_msg_channel_init_registered_srvid(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_msg_channel_init_state *state = tevent_req_data(
+ req, struct ctdb_msg_channel_init_state);
+ struct ctdb_reply_control *reply;
+ int ret;
+
+ ret = ctdb_conn_control_recv(subreq, talloc_tos(), &reply);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ if (reply->status != 0) {
+ tevent_req_error(req, EIO);
+ return;
+ }
+ state->channel = talloc(state, struct ctdb_msg_channel);
+ if (tevent_req_nomem(state->channel, req)) {
+ return;
+ }
+ state->channel->conn = talloc_move(state->channel, &state->conn);
+ tevent_req_done(req);
+}
+
+int ctdb_msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct ctdb_msg_channel **pchannel)
+{
+ struct ctdb_msg_channel_init_state *state = tevent_req_data(
+ req, struct ctdb_msg_channel_init_state);
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ *pchannel = talloc_move(mem_ctx, &state->channel);
+ return 0;
+}
+
+struct ctdb_msg_read_state {
+ size_t buflen;
+ uint8_t *buf;
+};
+
+static void ctdb_msg_channel_got_msg(struct tevent_req *subreq);
+
+struct tevent_req *ctdb_msg_read_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct ctdb_msg_channel *channel)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_msg_read_state *state;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct ctdb_msg_read_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ subreq = read_packet_send(state, ev, channel->conn->fd,
+ sizeof(uint32_t), ctdb_packet_more, NULL);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_msg_channel_got_msg, req);
+ return req;
+}
+
+static void ctdb_msg_channel_got_msg(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_msg_read_state *state = tevent_req_data(
+ req, struct ctdb_msg_read_state);
+ ssize_t nread;
+ uint8_t *buf;
+ int err;
+
+ nread = read_packet_recv(subreq, state, &buf, &err);
+ if (nread == -1) {
+ tevent_req_error(req, err);
+ return;
+ }
+ state->buflen = nread;
+ state->buf = buf;
+ tevent_req_done(req);
+}
+
+int ctdb_msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ uint8_t **pmsg, size_t *pmsg_len)
+{
+ struct ctdb_msg_read_state *state = tevent_req_data(
+ req, struct ctdb_msg_read_state);
+ struct ctdb_req_header *hdr;
+ struct ctdb_req_message *msg;
+ uint8_t *buf;
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+
+ hdr = (struct ctdb_req_header *)state->buf;
+ if (hdr->length != state->buflen) {
+ DEBUG(10, ("Got invalid header length\n"));
+ return EIO;
+ }
+ if (hdr->operation != CTDB_REQ_MESSAGE) {
+ DEBUG(10, ("Expected %d (CTDB_REQ_MESSAGE), got %d\n",
+ CTDB_REQ_MESSAGE, (int)hdr->operation));
+ return EIO;
+ }
+ if (hdr->length < offsetof(struct ctdb_req_message, data)) {
+ DEBUG(10, ("Got short msg, len=%d\n", (int)hdr->length));
+ return EIO;
+ }
+
+ msg = (struct ctdb_req_message *)hdr;
+ if (msg->datalen >
+ hdr->length - offsetof(struct ctdb_req_message, data)) {
+ DEBUG(10, ("Got invalid datalen %d\n", (int)msg->datalen));
+ return EIO;
+ }
+
+ buf = (uint8_t *)talloc_memdup(mem_ctx, msg->data, msg->datalen);
+ if (buf == NULL) {
+ return ENOMEM;
+ }
+ *pmsg = buf;
+ *pmsg_len = msg->datalen;
+ return 0;
+}
+
+#else
+
+struct dummy_state {
+ uint8_t dummy;
+};
+
+static struct tevent_req *dummy_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev)
+{
+ struct tevent_req *req;
+ struct dummy_state *state;
+ req = tevent_req_create(mem_ctx, &state, struct dummy_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ tevent_req_done(req);
+ return tevent_req_post(req, ev);
+}
+
+struct tevent_req *ctdb_conn_init_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ const char *sock)
+{
+ return dummy_send(mem_ctx, ev);
+}
+
+int ctdb_conn_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct ctdb_conn **pconn)
+{
+ return ENOSYS;
+}
+
+struct tevent_req *ctdb_conn_msg_write_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct ctdb_conn *conn,
+ uint32_t vnn, uint64_t srvid,
+ uint8_t *msg, size_t msg_len)
+{
+ return dummy_send(mem_ctx, ev);
+}
+
+int ctdb_conn_msg_write_recv(struct tevent_req *req)
+{
+ return ENOSYS;
+}
+
+struct tevent_req *ctdb_msg_channel_init_send(
+ TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+ const char *sock, uint64_t srvid)
+{
+ return dummy_send(mem_ctx, ev);
+}
+
+int ctdb_msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct ctdb_msg_channel **pchannel)
+{
+ return ENOSYS;
+}
+
+struct tevent_req *ctdb_msg_read_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct ctdb_msg_channel *channel)
+{
+ return dummy_send(mem_ctx, ev);
+}
+
+int ctdb_msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ uint8_t **pmsg, size_t *pmsg_len)
+{
+ return ENOSYS;
+}
+
+#endif
--- /dev/null
+/*
+ Unix SMB/CIFS implementation.
+ Test new ctdb API
+ Copyright (C) Volker Lendecke 2012
+
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ This program 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "includes.h"
+#include "torture/proto.h"
+
+#ifdef CLUSTER_SUPPORT
+
+#include "ctdb_conn.h"
+#include "lib/util/tevent_unix.h"
+#include "tdb.h"
+#include "ctdb_protocol.h"
+#include "messages.h"
+
+struct ctdb_conn_test_state {
+ struct tevent_context *ev;
+ struct ctdb_conn *conn;
+ struct ctdb_msg_channel *channel;
+ int msgno;
+};
+
+static void ctdb_conn_test_got_conn(struct tevent_req *subreq);
+static void ctdb_conn_test_got_pnn(struct tevent_req *subreq);
+static void ctdb_conn_test_got_channel(struct tevent_req *subreq);
+static void ctdb_conn_test_got_msg(struct tevent_req *subreq);
+static void ctdb_conn_test_msg_sent(struct tevent_req *subreq);
+
+static struct tevent_req *ctdb_conn_test_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev)
+{
+ struct tevent_req *req, *subreq;
+ struct ctdb_conn_test_state *state;
+
+ req = tevent_req_create(mem_ctx, &state, struct ctdb_conn_test_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->ev = ev;
+
+ subreq = ctdb_conn_init_send(mem_ctx, ev, lp_ctdbd_socket());
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_got_conn, req);
+ return req;
+}
+
+static void ctdb_conn_test_got_conn(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_test_state *state = tevent_req_data(
+ req, struct ctdb_conn_test_state);
+ uint64_t ret;
+
+ ret = ctdb_conn_init_recv(subreq, state, &state->conn);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ subreq = ctdb_conn_control_send(state, state->ev, state->conn,
+ CTDB_CURRENT_NODE,
+ CTDB_CONTROL_GET_PNN, 0, 0, NULL, 0);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_got_pnn, req);
+}
+
+static void ctdb_conn_test_got_pnn(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_test_state *state = tevent_req_data(
+ req, struct ctdb_conn_test_state);
+ int ret;
+ struct ctdb_reply_control *reply;
+
+ ret = ctdb_conn_control_recv(subreq, talloc_tos(), &reply);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ printf("vnn=%d\n", (int)reply->status);
+
+ subreq = ctdb_msg_channel_init_send(
+ state, state->ev, lp_ctdbd_socket(), 999999);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_got_channel, req);
+}
+
+static void ctdb_conn_test_got_channel(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_test_state *state = tevent_req_data(
+ req, struct ctdb_conn_test_state);
+ int ret;
+
+ ret = ctdb_msg_channel_init_recv(subreq, state, &state->channel);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+
+ subreq = ctdb_msg_read_send(state, state->ev, state->channel);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_got_msg, req);
+
+ state->msgno += 1;
+
+ subreq = ctdb_conn_msg_write_send(
+ state, state->ev, state->conn, CTDB_CURRENT_NODE, 999999,
+ (uint8_t *)&state->msgno, sizeof(state->msgno));
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_msg_sent, req);
+}
+
+static void ctdb_conn_test_got_msg(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_test_state *state = tevent_req_data(
+ req, struct ctdb_conn_test_state);
+ uint8_t *buf;
+ size_t buf_len;
+ int ret;
+
+ ret = ctdb_msg_read_recv(subreq, talloc_tos(), &buf, &buf_len);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ if (buf_len != sizeof(int)) {
+ printf("got invalid msg\n");
+ tevent_req_error(req, EINVAL);
+ return;
+ }
+ memcpy(&ret, buf, buf_len);
+ printf("got msg %d\n", ret);
+ if (ret == 5) {
+ tevent_req_done(req);
+ return;
+ }
+
+ subreq = ctdb_msg_read_send(state, state->ev, state->channel);
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_got_msg, req);
+}
+
+static void ctdb_conn_test_msg_sent(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct ctdb_conn_test_state *state = tevent_req_data(
+ req, struct ctdb_conn_test_state);
+ int ret;
+
+ ret = ctdb_conn_msg_write_recv(subreq);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ state->msgno += 1;
+
+ if (state->msgno >= 10) {
+ return;
+ }
+
+ subreq = ctdb_conn_msg_write_send(
+ state, state->ev, state->conn, CTDB_CURRENT_NODE, 999999,
+ (uint8_t *)&state->msgno, sizeof(state->msgno));
+ if (tevent_req_nomem(subreq, req)) {
+ return;
+ }
+ tevent_req_set_callback(subreq, ctdb_conn_test_msg_sent, req);
+}
+
+static int ctdb_conn_test_recv(struct tevent_req *req)
+{
+ int err;
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ return 0;
+}
+
+bool run_ctdb_conn(int dummy)
+{
+ struct tevent_context *ev;
+ struct tevent_req *req;
+ int ret;
+
+ ev = tevent_context_init(talloc_tos());
+ if (ev == NULL) {
+ fprintf(stderr, "tevent_context_init failed\n");
+ return false;
+ }
+ req = ctdb_conn_test_send(ev, ev);
+ if (req == NULL) {
+ fprintf(stderr, "ctdb_conn_test_send failed\n");
+ return false;
+ }
+ if (!tevent_req_poll(req, ev)) {
+ fprintf(stderr, "tevent_req_poll failed\n");
+ return false;
+ }
+ ret = ctdb_conn_test_recv(req);
+ TALLOC_FREE(req);
+ printf("ctdb_conn_test returned %s\n",
+ ret ? strerror(ret) : "success");
+ TALLOC_FREE(ev);
+ return (ret == 0);
+}
+
+#else /* CLUSTER_SUPPORT */
+
+bool run_ctdb_conn(int dummy)
+{
+ return true;
+}
+
+#endif
git.samba.org / kai / samba.git / commitdiff