This patch moves CLD from using manual data serialization to using XDR (via
rpcgen). Both the packet header and the message body are now serialized and
deserialized using XDR. This makes it easy to have a variable-length packet
header, as well as a variable-length message body.
This patch introduces a minor libcldc API change in struct cldc_call_opts.
v2 changes:
* Added __cld prefix to functions in cld_fmt.c
* When decoding CMT_GET messages, we should store the payload in the
session structure, rather than in cldc_call_opts.
* Got rid of pkt_is_first, pkt_is_last, in favor of a flags-based approach.
* Killed CLD_MAX_PKT_MSG. It's more efficient to only allocate as much space
as you need, rather than always allocating space for 128 packets in a message.
* Created CLD_MAX_PAYLOAD_SZ to mean the maximum size of the data that can be
sent with GET or PUT. This is different (and smaller than!) the maximum
message size.
* automake: Add cld_msg_rpc.h to BUILT_SOURCES
* automake: Add lib dir to INCLUDES
* automake: "make clean" now deletes XDR build products
Signed-off-by: Colin McCabe <cmccabe@xxxxxxxxxxxxxx>
---
.gitignore | 4 +
include/Makefile.am | 2 +-
include/cld_common.h | 10 +
include/cld_fmt.h | 89 +++++
include/cld_msg.h | 252 -------------
include/cldc.h | 36 +-
lib/Makefile.am | 19 +-
lib/cld_fmt.c | 193 ++++++++++
lib/cld_msg_rpc.x | 218 +++++++++++
lib/cldc.c | 966 +++++++++++++++++++++---------------------------
lib/common.c | 6 +-
server/Makefile.am | 9 +-
server/cld.h | 97 ++++--
server/cldb.h | 2 +-
server/msg.c | 304 ++++++----------
server/server.c | 743 ++++++++++++++++++++-----------------
server/session.c | 396 ++++++++++-----------
test/Makefile.am | 5 +-
test/load-file-event.c | 14 +-
test/save-file-event.c | 2 -
tools/cldcli.c | 22 +-
21 files changed, 1777 insertions(+), 1612 deletions(-)
create mode 100644 include/cld_fmt.h
delete mode 100644 include/cld_msg.h
create mode 100644 lib/cld_fmt.c
create mode 100644 lib/cld_msg_rpc.x
diff --git a/.gitignore b/.gitignore
index 79a4cd3..805fa22 100644
--- a/.gitignore
+++ b/.gitignore
@@ -32,5 +32,9 @@ cld-config.h*
cscope.*
ncscope.*
+# XDR files
+*_rpc.h
+*_rpc_xdr.c
+
# ignore Doxygen output directory
gendoc
diff --git a/include/Makefile.am b/include/Makefile.am
index 888088c..f90d222 100644
--- a/include/Makefile.am
+++ b/include/Makefile.am
@@ -1,5 +1,5 @@
EXTRA_DIST = cld-private.h libtimer.h
-include_HEADERS = cldc.h cld_msg.h hail_log.h cld_common.h
+include_HEADERS = cldc.h hail_log.h cld_common.h cld_fmt.h
diff --git a/include/cld_common.h b/include/cld_common.h
index 116ef93..217d1ad 100644
--- a/include/cld_common.h
+++ b/include/cld_common.h
@@ -20,10 +20,20 @@
*/
#include <stdint.h>
+#include <lib/cld_msg_rpc.h>
+
+#define CLD_ALIGN8(n) ((8 - ((n) & 7)) & 7)
unsigned long long cld_sid2llu(const uint8_t *sid);
void __cld_rand64(void *p);
const char *cld_errstr(enum cle_err_codes ecode);
int cld_readport(const char *fname);
+/*
+ * * We use a unified format for sid so it can be searched in log files (* in
+ * vi).
+ */
+#define SIDFMT "%016llX"
+#define SIDARG(sid) cld_sid2llu(sid)
+
#endif /* __CLD_COMMON_H__ */
diff --git a/include/cld_fmt.h b/include/cld_fmt.h
new file mode 100644
index 0000000..7af84a8
--- /dev/null
+++ b/include/cld_fmt.h
@@ -0,0 +1,89 @@
+#ifndef __CLD_FMT_H__
+#define __CLD_FMT_H__
+
+/*
+ * Copyright 2010, Colin McCabe
+ *
+ * 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.
+ *
+ * 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; see the file COPYING. If not, write to
+ * the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <openssl/sha.h>
+#include <lib/cld_msg_rpc.h>
+#include <stdbool.h>
+
+/* @file cld_fmt.h
+ *
+ * This file has some definitions and helper functions pertaining to the CLD
+ * network protocol. Unlike cld_msg.x, it's not an XDR file.
+ */
+
+struct hail_log;
+
+/*** Validate the HMAC signature of a byte buffer.
+ *
+ * @param log log to write to
+ * @param key The key, as a NULL-terminated string
+ * @param buf The buffer
+ * @param buf_len Length of the buffer
+ * @param sha The signature itself. Must be of length exactly
+ * SHA_DIGEST_LENGTH
+ *
+ * @return 0 on success; error code otherwise
+ */
+int __cld_authcheck(struct hail_log *log, const char *key,
+ const char *buf, size_t buf_len, const char *sha);
+
+/*** Sign a byte buffer.
+ *
+ * @param log log to write to
+ * @param key The key, as a NULL-terminated string
+ * @param buf The buffer
+ * @param buf_len Length of the buffer
+ * @param sha (out param) The signature itself. Must be of length
+ * exactly SHA_DIGEST_LENGTH
+ *
+ * @return 0 on success; error code otherwise
+ */
+int __cld_authsign(struct hail_log *log, const char *key,
+ const char *buf, size_t buf_len, char *sha);
+
+/* Returns a constant string representing a message type */
+const char *__cld_msg_type_to_str(enum cld_msg_type type);
+
+/* Returns a string representation of a packet header
+ *
+ * @param scratch (out param) buffer of length
+ * PKT_HDR_TO_STR_SCRATCH_LEN
+ * @param pkt_hdr packet header
+ * @param pkt_len length of packet
+ *
+ * @return pointer to 'scratch'
+ */
+const char *__cld_pkt_hdr_to_str(char *scratch,
+ const char *pkt_hdr, size_t pkt_len);
+
+void __cld_dump_buf(const void *buf, size_t len);
+
+/** Footer that appears at the end of each packet */
+struct __attribute__((packed)) cld_pkt_ftr {
+ uint64_t seqid; /**< packet sequence ID */
+ char sha[SHA_DIGEST_LENGTH]; /**< packet signature */
+};
+
+/** Length of the packet footer. This size is fixed */
+#define CLD_PKT_FTR_LEN sizeof(struct cld_pkt_ftr)
+
+#define PKT_HDR_TO_STR_SCRATCH_LEN 128
+
+#endif
diff --git a/include/cld_msg.h b/include/cld_msg.h
deleted file mode 100644
index 513583d..0000000
--- a/include/cld_msg.h
+++ /dev/null
@@ -1,252 +0,0 @@
-#ifndef __CLD_MSG_H__
-#define __CLD_MSG_H__
-
-/*
- * Copyright 2009 Red Hat, Inc.
- *
- * 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.
- *
- * 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; see the file COPYING. If not, write to
- * the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
- *
- */
-
-#include <stdint.h>
-
-#define CLD_PKT_MAGIC "CLDc1pkt"
-#define CLD_MSG_MAGIC "CLDc1msg"
-
-#define CLD_ALIGN8(n) ((8 - ((n) & 7)) & 7)
-
-enum {
- CLD_MAGIC_SZ = 8, /**< length of magic number */
- CLD_SID_SZ = 8, /**< length of session id */
-
- CLD_INODE_NAME_MAX = 256, /**< max total pathname len */
-
- CLD_MAX_USERNAME = 32, /**< includes req. nul */
- CLD_MAX_SECRET_KEY = 128, /**< includes req. nul */
-
- CLD_MAX_PKT_MSG_SZ = 1024,
- CLD_MAX_PKT_MSG = 128,
- CLD_MAX_MSG_SZ = CLD_MAX_PKT_MSG * 1024, /**< maximum total
- msg size, including all packets */
-};
-
-/*
- * We use a unified format for sid so it can be searched in log files (* in vi).
- */
-#define SIDFMT "%016llX"
-#define SIDARG(sid) cld_sid2llu(sid)
-
-/** available RPC operations */
-enum cld_msg_ops {
- /* client -> server */
- cmo_nop = 0, /**< no op */
- cmo_new_sess = 1, /**< new session */
- cmo_open = 2, /**< open file */
- cmo_get_meta = 3, /**< get metadata */
- cmo_get = 4, /**< get metadata + data */
- cmo_put = 6, /**< put data */
- cmo_close = 7, /**< close file */
- cmo_del = 8, /**< delete file */
- cmo_lock = 9, /**< lock */
- cmo_unlock = 10, /**< unlock */
- cmo_trylock = 11, /**< trylock */
- cmo_ack = 12, /**< ack of seqid rx'd */
- cmo_end_sess = 13, /**< end session */
-
- /* server -> client */
- cmo_ping = 30, /**< server to client ping */
- cmo_not_master = 31, /**< I am not the master! */
- cmo_event = 32, /**< server->cli async event */
- cmo_ack_frag = 33, /**< ack partial msg */
-};
-
-/** CLD error codes */
-enum cle_err_codes {
- CLE_OK = 0, /**< success / no error */
- CLE_SESS_EXISTS = 1, /**< session exists */
- CLE_SESS_INVAL = 2, /**< session doesn't exist */
- CLE_DB_ERR = 3, /**< db error */
- CLE_BAD_PKT = 4, /**< invalid/corrupted packet */
- CLE_INODE_INVAL = 5, /**< inode doesn't exist */
- CLE_NAME_INVAL = 6, /**< inode name invalid */
- CLE_OOM = 7, /**< server out of memory */
- CLE_FH_INVAL = 8, /**< file handle invalid */
- CLE_DATA_INVAL = 9, /**< invalid data pkt */
- CLE_LOCK_INVAL = 10, /**< invalid lock */
- CLE_LOCK_CONFLICT = 11, /**< conflicting lock held */
- CLE_LOCK_PENDING = 12, /**< lock waiting to be acq. */
- CLE_MODE_INVAL = 13, /**< op incompat. w/ file mode */
- CLE_INODE_EXISTS = 14, /**< inode exists */
- CLE_DIR_NOTEMPTY = 15, /**< dir not empty */
- CLE_INTERNAL_ERR = 16, /**< nonspecific internal err */
- CLE_TIMEOUT = 17, /**< session timed out */
- CLE_SIG_INVAL = 18, /**< HMAC sig bad / auth failed */
-};
-
-/** availble OPEN mode flags */
-enum cld_open_modes {
- COM_READ = (1 << 0), /**< read */
- COM_WRITE = (1 << 1), /**< write */
- COM_LOCK = (1 << 2), /**< lock */
- COM_ACL = (1 << 3), /**< ACL update */
- COM_CREATE = (1 << 4), /**< create file, if not exist */
- COM_EXCL = (1 << 5), /**< fail create if file exists */
- COM_DIRECTORY = (1 << 6), /**< operate on a directory */
-};
-
-/** potential events client may receive */
-enum cld_events {
- CE_UPDATED = (1 << 0), /**< contents updated */
- CE_DELETED = (1 << 1), /**< inode deleted */
- CE_LOCKED = (1 << 2), /**< lock acquired */
- CE_MASTER_FAILOVER = (1 << 3), /**< master failover */
- CE_SESS_FAILED = (1 << 4),
-};
-
-/** LOCK flags */
-enum cld_lock_flags {
- CLF_SHARED = (1 << 0), /**< a shared (read) lock */
-};
-
-/** CLD packet flags */
-enum cld_packet_flags {
- CPF_FIRST = (1 << 0), /**< first fragment */
- CPF_LAST = (1 << 1), /**< last fragment */
-};
-
-/** header for each packet */
-struct cld_packet {
- uint8_t magic[CLD_MAGIC_SZ]; /**< magic number; constant */
- uint64_t seqid; /**< sequence id */
- uint8_t sid[CLD_SID_SZ]; /**< client id */
- uint32_t flags; /**< CPF_xxx flags */
- uint8_t res[4];
- char user[CLD_MAX_USERNAME]; /**< authenticated user */
-};
-
-/** header for each message */
-struct cld_msg_hdr {
- uint8_t magic[CLD_MAGIC_SZ]; /**< magic number; constant */
- uint64_t xid; /**< opaque message id */
- uint8_t op; /**< operation code */
- uint8_t res1[7];
-};
-
-/** standard response for each message */
-struct cld_msg_resp {
- struct cld_msg_hdr hdr;
-
- uint32_t code; /**< error code, CLE_xxx */
- uint32_t rsv; /**< reserved */
- uint64_t xid_in; /**< C->S xid */
-};
-
-/** ACK-FRAG message */
-struct cld_msg_ack_frag {
- struct cld_msg_hdr hdr;
-
- uint64_t seqid; /**< sequence id to ack */
-};
-
-/** OPEN message */
-struct cld_msg_open {
- struct cld_msg_hdr hdr;
-
- uint32_t mode; /**< open mode, COM_xxx */
- uint32_t events; /**< events mask, CE_xxx */
- uint16_t name_len; /**< length of file name */
- uint8_t res[6];
- /* inode name */
-};
-
-/** OPEN message response */
-struct cld_msg_open_resp {
- struct cld_msg_resp resp;
-
- uint64_t fh; /**< handle opened */
-};
-
-/** GET message */
-struct cld_msg_get {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
-};
-
-/** GET message response */
-struct cld_msg_get_resp {
- struct cld_msg_resp resp;
-
- uint64_t inum; /**< unique inode number */
- uint32_t ino_len; /**< inode name len */
- uint32_t size; /**< data size */
- uint64_t version; /**< inode version */
- uint64_t time_create; /**< creation time */
- uint64_t time_modify; /**< last modification time */
- uint32_t flags; /**< inode flags; CIFL_xxx */
- uint8_t res[4];
- /* inode name */
-};
-
-/** PUT message */
-struct cld_msg_put {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
- uint32_t data_size; /**< total size of data */
- uint8_t res[4];
-};
-
-/** CLOSE message */
-struct cld_msg_close {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
-};
-
-/** DEL message */
-struct cld_msg_del {
- struct cld_msg_hdr hdr;
-
- uint16_t name_len; /**< length of file name */
- uint8_t res[6];
- /* inode name */
-};
-
-/** UNLOCK message */
-struct cld_msg_unlock {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
-};
-
-/** LOCK message */
-struct cld_msg_lock {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
- uint32_t flags; /**< CLF_xxx */
- uint8_t res[4];
-};
-
-/** Server-to-client EVENT message */
-struct cld_msg_event {
- struct cld_msg_hdr hdr;
-
- uint64_t fh; /**< open file handle */
- uint32_t events; /**< CE_xxx */
- uint8_t res[4];
-};
-
-#endif /* __CLD_MSG_H__ */
diff --git a/include/cldc.h b/include/cldc.h
index f1db7d2..330adb0 100644
--- a/include/cldc.h
+++ b/include/cldc.h
@@ -19,10 +19,11 @@
*
*/
+
#include <sys/types.h>
#include <stdbool.h>
#include <glib.h>
-#include <cld_msg.h>
+#include <lib/cld_msg_rpc.h>
#include <cld_common.h>
#include <hail_log.h>
@@ -35,33 +36,30 @@ struct cldc_call_opts {
void *private;
/* private; lib-owned */
- enum cld_msg_ops op;
- union {
- struct {
- struct cld_msg_get_resp resp;
- const char *buf;
- unsigned int size;
- char inode_name[CLD_INODE_NAME_MAX];
- } get;
- } u;
+ struct cld_msg_get_resp resp;
};
+void cldc_call_opts_get_data(struct cldc_call_opts *copts,
+ char **data, size_t *data_len);
+
struct cldc_pkt_info {
int pkt_len;
+ int hdr_len;
int retries;
+ char user[CLD_MAX_USERNAME];
/* must be at end of struct */
- struct cld_packet pkt;
- uint8_t data[0];
+ char data[0];
};
/** an outgoing message, from client to server */
struct cldc_msg {
uint64_t xid;
-
+ enum cld_msg_type type;
struct cldc_session *sess;
- ssize_t (*cb)(struct cldc_msg *, const void *, size_t, bool);
+ ssize_t (*cb)(struct cldc_msg *, const void *, size_t,
+ enum cle_err_codes resp_rc);
void *cb_private;
struct cldc_call_opts copts;
@@ -70,18 +68,15 @@ struct cldc_msg {
time_t expire_time;
- int data_len;
int n_pkts;
- struct cldc_pkt_info *pkt_info[CLD_MAX_PKT_MSG];
-
/* must be at end of struct */
- uint8_t data[0];
+ struct cldc_pkt_info *pkt_info[0];
};
/** an open file handle associated with a session */
struct cldc_fh {
- uint64_t fh_le; /* fh id, LE */
+ uint64_t fh;
struct cldc_session *sess;
bool valid;
};
@@ -128,8 +123,11 @@ struct cldc_session {
bool confirmed;
+ enum cld_msg_type msg_buf_type;
unsigned int msg_buf_len;
char msg_buf[CLD_MAX_MSG_SZ];
+ char payload[CLD_MAX_PAYLOAD_SZ];
+ char inode_name_temp[CLD_INODE_NAME_MAX];
};
/** Information for a single CLD server host */
diff --git a/lib/Makefile.am b/lib/Makefile.am
index 68be429..6275ca6 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -1,15 +1,31 @@
+BUILT_SOURCES = cld_msg_rpc.h
+
EXTRA_DIST =
libcldc.pc.in libcldc-uninstalled.pc.in
INCLUDES = -I$(top_srcdir)/include \
@GLIB_CFLAGS@
+mostlyclean-local:
+ -rm -f *_rpc.h *_rpc_xdr.c
+
+%_rpc.h: %_rpc.x
+ rpcgen -h $< > $@
+
+%_rpc_xdr.c: %_rpc.x
+ rpcgen -c $< > $@
+
LINK = $(LIBTOOL) --mode=link $(CC) $(CFLAGS) $(LDFLAGS) -o $@
lib_LTLIBRARIES = libcldc.la
-libcldc_la_SOURCES = cldc.c cldc-udp.c cldc-dns.c common.c
+libcldc_la_SOURCES = cldc.c \
+ cldc-udp.c \
+ cldc-dns.c \
+ common.c \
+ cld_fmt.c \
+ cld_msg_rpc_xdr.c
libcldc_la_LDFLAGS = \
-version-info $(LIBCLDC_CURRENT):$(LIBCLDC_REVISION):$(LIBCLDC_AGE) \
@@ -21,3 +37,4 @@ noinst_LIBRARIES = libtimer.a
pkgconfigdir = $(libdir)/pkgconfig
pkgconfig_DATA = libcldc.pc
+include_HEADERS = cld_msg_rpc.h
diff --git a/lib/cld_fmt.c b/lib/cld_fmt.c
new file mode 100644
index 0000000..607a035
--- /dev/null
+++ b/lib/cld_fmt.c
@@ -0,0 +1,193 @@
+/*
+ * Copyright 2010, Colin McCabe
+ *
+ * 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.
+ *
+ * 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; see the file COPYING. If not, write to
+ * the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+#include <stdlib.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <errno.h>
+#include <glib.h>
+#include <cld-private.h>
+#include <openssl/sha.h>
+#include <openssl/hmac.h>
+#include "cld_fmt.h"
+#include "cld_msg_rpc.h"
+#include <hail_log.h>
+#include <syslog.h>
+
+int __cld_authcheck(struct hail_log *log, const char *key,
+ const char *buf, size_t buf_len,
+ const char *sha)
+{
+ unsigned char md[SHA_DIGEST_LENGTH];
+ unsigned int md_len = 0;
+
+ if (!key[0])
+ return EINVAL;
+ HMAC(EVP_sha1(), key, strlen(key),
+ (uint8_t *)buf, buf_len, md, &md_len);
+
+ if (md_len != SHA_DIGEST_LENGTH) {
+ HAIL_CRIT(log, "__cld_authsign BUG: md_len != "
+ "SHA_DIGEST_LENGTH");
+ return EBADMSG; /* BUG */
+ }
+
+ if (memcmp(md, sha, SHA_DIGEST_LENGTH))
+ return EPERM;
+
+ return 0;
+}
+
+int __cld_authsign(struct hail_log *log, const char *key,
+ const char *buf, size_t buf_len,
+ char *sha)
+{
+ unsigned char md[SHA_DIGEST_LENGTH];
+ unsigned int md_len = 0;
+
+ if (!key[0]) {
+ HAIL_DEBUG(log, "%s: invalid key\n", __func__);
+ return EINVAL;
+ }
+
+ HMAC(EVP_sha1(), key, strlen(key),
+ (uint8_t *)buf, buf_len, md, &md_len);
+
+ if (md_len != SHA_DIGEST_LENGTH) {
+ HAIL_CRIT(log, "__cld_authsign BUG: md_len != "
+ "SHA_DIGEST_LENGTH");
+ return EBADMSG;
+ }
+
+ memcpy(sha, md, SHA_DIGEST_LENGTH);
+ return 0;
+}
+
+const char *__cld_msg_type_to_str(enum cld_msg_type type)
+{
+ switch (type) {
+ case CMT_NOP: return "CMT_NOP";
+ case CMT_NEW_SESS: return "CMT_NEW_SESS";
+ case CMT_OPEN: return "CMT_OPEN";
+ case CMT_GET_META: return "CMT_GET_META";
+ case CMT_GET: return "CMT_GET";
+ case CMT_PUT: return "CMT_PUT";
+ case CMT_CLOSE: return "CMT_CLOSE";
+ case CMT_DEL: return "CMT_DEL";
+ case CMT_LOCK: return "CMT_LOCK";
+ case CMT_UNLOCK: return "CMT_UNLOCK";
+ case CMT_TRYLOCK: return "CMT_TRYLOCK";
+ case CMT_ACK: return "CMT_ACK";
+ case CMT_END_SESS: return "CMT_END_SESS";
+ case CMT_PING: return "CMT_PING";
+ case CMT_NOT_MASTER: return "CMT_NOT_MASTER";
+ case CMT_EVENT: return "CMT_EVENT";
+ case CMT_ACK_FRAG: return "CMT_ACK_FRAG";
+ default: return "(unknown)";
+ }
+}
+
+const char *__cld_pkt_hdr_to_str(char *scratch,
+ const char *pkt_hdr, size_t pkt_len)
+{
+ XDR xin;
+ struct cld_pkt_hdr pkt;
+ bool bad_magic;
+ char temp[50], temp2[50];
+ uint64_t seqid;
+ struct cld_pkt_ftr *foot;
+ size_t hdr_len;
+
+ temp[0] = '\0';
+ temp2[0] = '\0';
+ foot = (struct cld_pkt_ftr *)(pkt_hdr + pkt_len - CLD_PKT_FTR_LEN);
+ seqid = le64_to_cpu(foot->seqid);
+
+ if (pkt_len <= CLD_PKT_FTR_LEN) {
+ snprintf(scratch, PKT_HDR_TO_STR_SCRATCH_LEN,
+ "[MALFORMED: only %d bytes]", pkt_len);
+ return scratch;
+ }
+ xdrmem_create(&xin, (void *)pkt_hdr, pkt_len - CLD_PKT_FTR_LEN,
+ XDR_DECODE);
+ memset(&pkt, 0, sizeof(pkt));
+ if (!xdr_cld_pkt_hdr(&xin, &pkt)) {
+ xdr_destroy(&xin);
+ snprintf(scratch, PKT_HDR_TO_STR_SCRATCH_LEN,
+ "[MALFORMED: can't parse]");
+ return scratch;
+ }
+ hdr_len = xdr_getpos(&xin);
+ xdr_destroy(&xin);
+
+ bad_magic = !!(memcmp(&pkt.magic, CLD_PKT_MAGIC, sizeof(pkt.magic)));
+ if (pkt.mi.order & CLD_PKT_IS_FIRST) {
+ struct cld_pkt_msg_infos *infos =
+ &pkt.mi.cld_pkt_msg_info_u.mi;
+ snprintf(temp, sizeof(temp), "[TYPE:%s, XID:%llx]",
+ __cld_msg_type_to_str(infos->type), infos->xid);
+ switch (infos->type) {
+ case CMT_ACK_FRAG: {
+ XDR x;
+ struct cld_msg_ack_frag ack;
+ memset(&ack, 0, sizeof(ack));
+ xdrmem_create(&x, ((char *)pkt_hdr) + hdr_len,
+ pkt_len - hdr_len - CLD_PKT_FTR_LEN,
+ XDR_DECODE);
+ if (!xdr_cld_msg_ack_frag(&x, &ack)) {
+ xdr_destroy(&x);
+ snprintf(temp2, sizeof(temp2), "{MALFORMED}");
+ break;
+ }
+ snprintf(temp2, sizeof(temp2), "{seqid:%llx}",
+ ack.seqid);
+ xdr_destroy(&x);
+ break;
+ }
+ default:
+ break;
+ }
+ } else {
+ snprintf(temp, sizeof(temp), "[CONT]");
+ }
+
+ snprintf(scratch, PKT_HDR_TO_STR_SCRATCH_LEN,
+ "<%s%s%s> "
+ "%s USER:'%s' SEQID:%llu %s",
+ ((pkt.mi.order & CLD_PKT_IS_FIRST) ? "1st" : ""),
+ ((pkt.mi.order & CLD_PKT_IS_LAST) ? "End" : ""),
+ (bad_magic ? "B" : ""),
+ temp, pkt.user, seqid, temp2);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return scratch;
+}
+
+void __cld_dump_buf(const void *buf, size_t len)
+{
+ const unsigned char *buff = buf;
+ size_t off = 0;
+ do {
+ int i;
+ for (i = 0; i < 8; i++) {
+ if (!len)
+ break;
+ printf("%02x ", buff[off++]);
+ len--;
+ }
+ printf("\n");
+ } while (len);
+}
diff --git a/lib/cld_msg_rpc.x b/lib/cld_msg_rpc.x
new file mode 100644
index 0000000..a819ffb
--- /dev/null
+++ b/lib/cld_msg_rpc.x
@@ -0,0 +1,218 @@
+/*
+ * Copyright 2010, Colin McCabe
+ *
+ * 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.
+ *
+ * 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; see the file COPYING. If not, write to
+ * the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
+const CLD_PKT_MAGIC = "CLDc1pkt";
+const CLD_SID_SZ = 8;
+
+const CLD_INODE_NAME_MAX = 256; /**< max total pathname len */
+
+const CLD_MAX_USERNAME = 32;
+
+const CLD_MAX_PKT_MSG_SZ = 1024; /**< The maximum number of message bytes we'll
+ put in a single packet */
+
+const CLD_MAX_PAYLOAD_SZ = 131072; /**< Maximum length of the data that can be
+ sent with get or put. In some sense,
+ this is part of cld's API, and
+ shouldn't be changed lightly. */
+
+const CLD_MAX_MSG_SZ = 196608; /**< Maximum size of a single message
+ including all packets. */
+
+const CLD_MAX_SECRET_KEY = 128; /**< includes req. nul */
+
+/** available RPC operations */
+enum cld_msg_type {
+ /* client -> server */
+ CMT_NOP = 0, /**< no op */
+ CMT_NEW_SESS = 1, /**< new session */
+ CMT_OPEN = 2, /**< open file */
+ CMT_GET_META = 3, /**< get metadata */
+ CMT_GET = 4, /**< get metadata + data */
+ CMT_PUT = 6, /**< put data */
+ CMT_CLOSE = 7, /**< close file */
+ CMT_DEL = 8, /**< delete file */
+ CMT_LOCK = 9, /**< lock */
+ CMT_UNLOCK = 10, /**< unlock */
+ CMT_TRYLOCK = 11, /**< trylock */
+ CMT_ACK = 12, /**< ack of seqid rx'd */
+ CMT_END_SESS = 13, /**< end session */
+
+ /* server -> client */
+ CMT_PING = 14, /**< server to client ping */
+ CMT_NOT_MASTER = 15, /**< I am not the master! */
+ CMT_EVENT = 16, /**< server->cli async event */
+ CMT_ACK_FRAG = 17, /**< ack partial msg */
+
+ CMT_AFTER_LAST
+};
+
+/** CLD error codes */
+enum cle_err_codes {
+ CLE_OK = 0, /**< success / no error */
+ CLE_SESS_EXISTS = 1, /**< session exists */
+ CLE_SESS_INVAL = 2, /**< session doesn't exist */
+ CLE_DB_ERR = 3, /**< db error */
+ CLE_BAD_PKT = 4, /**< invalid/corrupted packet */
+ CLE_INODE_INVAL = 5, /**< inode doesn't exist */
+ CLE_NAME_INVAL = 6, /**< inode name invalid */
+ CLE_OOM = 7, /**< server out of memory */
+ CLE_FH_INVAL = 8, /**< file handle invalid */
+ CLE_DATA_INVAL = 9, /**< invalid data pkt */
+ CLE_LOCK_INVAL = 10, /**< invalid lock */
+ CLE_LOCK_CONFLICT = 11, /**< conflicting lock held */
+ CLE_LOCK_PENDING = 12, /**< lock waiting to be acq. */
+ CLE_MODE_INVAL = 13, /**< op incompat. w/ file mode */
+ CLE_INODE_EXISTS = 14, /**< inode exists */
+ CLE_DIR_NOTEMPTY = 15, /**< dir not empty */
+ CLE_INTERNAL_ERR = 16, /**< nonspecific internal err */
+ CLE_TIMEOUT = 17, /**< session timed out */
+ CLE_SIG_INVAL = 18 /**< HMAC sig bad / auth failed */
+};
+
+/** availble OPEN mode flags */
+enum cld_open_modes {
+ COM_READ = 0x01, /**< read */
+ COM_WRITE = 0x02, /**< write */
+ COM_LOCK = 0x04, /**< lock */
+ COM_ACL = 0x08, /**< ACL update */
+ COM_CREATE = 0x10, /**< create file, if not exist */
+ COM_EXCL = 0x20, /**< fail create if file exists */
+ COM_DIRECTORY = 0x40 /**< operate on a directory */
+};
+
+/** potential events client may receive */
+enum cld_events {
+ CE_UPDATED = 0x01, /**< contents updated */
+ CE_DELETED = 0x02, /**< inode deleted */
+ CE_LOCKED = 0x04, /**< lock acquired */
+ CE_MASTER_FAILOVER = 0x08, /**< master failover */
+ CE_SESS_FAILED = 0x10
+};
+
+/** LOCK flags */
+enum cld_lock_flags {
+ CLF_SHARED = 0x01 /**< a shared (read) lock */
+};
+
+/** Describes whether a packet begins, continues, or ends a message. */
+enum cld_pkt_order_t {
+ CLD_PKT_ORD_MID = 0x0,
+ CLD_PKT_ORD_FIRST = 0x1,
+ CLD_PKT_ORD_LAST = 0x2,
+ CLD_PKT_ORD_FIRST_LAST = 0x3
+};
+const CLD_PKT_IS_FIRST = 0x1;
+const CLD_PKT_IS_LAST = 0x2;
+
+/** Information that appears only in the first packet */
+struct cld_pkt_msg_infos {
+ hyper xid; /**< opaque message id */
+ enum cld_msg_type type; /**< type of the message */
+};
+
+/** Information about the message contained in this packet */
+union cld_pkt_msg_info switch (enum cld_pkt_order_t order) {
+ case CLD_PKT_ORD_MID:
+ case CLD_PKT_ORD_LAST:
+ void;
+ case CLD_PKT_ORD_FIRST:
+ case CLD_PKT_ORD_FIRST_LAST:
+ struct cld_pkt_msg_infos mi;
+};
+
+/** header for each packet */
+struct cld_pkt_hdr {
+ hyper magic; /**< magic number; constant */
+ hyper sid; /**< client id */
+ string user<CLD_MAX_USERNAME>; /**< authenticated user */
+ struct cld_pkt_msg_info mi;
+};
+
+/** generic response for PUT, CLOSE, DEL, LOCK, UNLOCK */
+struct cld_msg_generic_resp {
+ enum cle_err_codes code; /**< error code, CLE_xxx */
+ hyper xid_in; /**< C->S xid */
+};
+
+/** ACK-FRAG message */
+struct cld_msg_ack_frag {
+ hyper seqid; /**< sequence id to ack */
+};
+
+/** OPEN message */
+struct cld_msg_open {
+ int mode; /**< open mode, COM_xxx */
+ int events; /**< events mask, CE_xxx */
+ string inode_name<CLD_INODE_NAME_MAX>;
+};
+
+/** OPEN message response */
+struct cld_msg_open_resp {
+ struct cld_msg_generic_resp msg;
+ hyper fh; /**< handle opened */
+};
+
+/** GET message */
+struct cld_msg_get {
+ hyper fh; /**< open file handle */
+};
+
+/** GET message response */
+struct cld_msg_get_resp {
+ struct cld_msg_generic_resp msg;
+ hyper inum; /**< unique inode number */
+ hyper vers; /**< inode version */
+ hyper time_create; /**< creation time */
+ hyper time_modify; /**< last modification time */
+ int flags; /**< inode flags; CIFL_xxx */
+ string inode_name<CLD_INODE_NAME_MAX>;
+ opaque data<CLD_MAX_PAYLOAD_SZ>;
+};
+
+/** PUT message */
+struct cld_msg_put {
+ hyper fh; /**< open file handle */
+ opaque data<CLD_MAX_PAYLOAD_SZ>;
+};
+
+/** CLOSE message */
+struct cld_msg_close {
+ hyper fh; /**< open file handle */
+};
+
+/** DEL message */
+struct cld_msg_del {
+ string inode_name<CLD_INODE_NAME_MAX>;
+};
+
+/** UNLOCK message */
+struct cld_msg_unlock {
+ uint64_t fh; /**< open file handle */
+};
+
+/** LOCK message */
+struct cld_msg_lock {
+ hyper fh; /**< open file handle */
+ int flags; /**< CLF_xxx */
+};
+
+/** Server-to-client EVENT message */
+struct cld_msg_event {
+ hyper fh; /**< open file handle */
+ int events; /**< CE_xxx */
+};
diff --git a/lib/cldc.c b/lib/cldc.c
index bc4b48c..603700c 100644
--- a/lib/cldc.c
+++ b/lib/cldc.c
@@ -35,6 +35,8 @@
#include <glib.h>
#include <cld-private.h>
#include <cldc.h>
+#include <cld_fmt.h>
+#include <lib/cld_msg_rpc.h>
#include <syslog.h>
enum {
@@ -45,14 +47,8 @@ enum {
CLDC_SESS_EXPIRE = 2 * 60,
};
-static bool authsign(struct cldc_session *, struct cld_packet *, size_t);
static int sess_send_pkt(struct cldc_session *sess,
- const struct cld_packet *pkt, size_t pkt_len);
-
-static const struct cld_msg_hdr def_msg_ack = {
- .magic = CLD_MSG_MAGIC,
- .op = cmo_ack,
-};
+ const void *pkt, size_t pkt_len);
#ifndef HAVE_STRNLEN
static size_t strnlen(const char *s, size_t maxlen)
@@ -80,6 +76,13 @@ static size_t strnlen(const char *s, size_t maxlen)
#define EBADE 52
#endif
+void cldc_call_opts_get_data(struct cldc_call_opts *copts,
+ char **data, size_t *data_len)
+{
+ *data = copts->resp.data.data_val;
+ *data_len = copts->resp.data.data_len;
+}
+
static void cldc_errlog(int prio, const char *fmt, ...)
{
char buf[200];
@@ -93,94 +96,128 @@ static void cldc_errlog(int prio, const char *fmt, ...)
static int ack_seqid(struct cldc_session *sess, uint64_t seqid_le)
{
- struct cld_packet *pkt;
- struct cld_msg_hdr *resp;
- size_t pkt_len;
-
- pkt_len = sizeof(*pkt) + sizeof(*resp) + SHA_DIGEST_LENGTH;
- pkt = alloca(pkt_len);
- memset(pkt, 0, pkt_len);
-
- memcpy(pkt->magic, CLD_PKT_MAGIC, CLD_MAGIC_SZ);
- pkt->seqid = seqid_le;
- memcpy(pkt->sid, sess->sid, CLD_SID_SZ);
- pkt->flags = cpu_to_le32(CPF_FIRST | CPF_LAST);
- strncpy(pkt->user, sess->user, CLD_MAX_USERNAME - 1);
-
- resp = (struct cld_msg_hdr *) (pkt + 1);
- memcpy(resp, &def_msg_ack, sizeof(*resp));
-
- if (!authsign(sess, pkt, pkt_len)) {
- HAIL_INFO(&sess->log, "authsign failed 2");
- return -1;
+ XDR xdrs;
+ size_t hdr_len, total_len;
+ char buf[CLD_MAX_PKT_MSG_SZ];
+ struct cld_pkt_hdr pkt;
+ struct cld_pkt_ftr *foot;
+ int ret;
+ static const char * const magic = CLD_PKT_MAGIC;
+
+ /* Construct ACK packet */
+ memset(&pkt, 0, sizeof(struct cld_pkt_hdr));
+ memcpy(&pkt.magic, magic, sizeof(pkt.magic));
+ memcpy(&pkt.sid, sess->sid, CLD_SID_SZ);
+ pkt.user = sess->user;
+ pkt.mi.order = CLD_PKT_ORD_FIRST_LAST;
+ pkt.mi.cld_pkt_msg_info_u.mi.xid = 0;
+ pkt.mi.cld_pkt_msg_info_u.mi.type = CMT_ACK;
+
+ /* Serialize packet */
+ xdrmem_create(&xdrs, (char *)buf,
+ sizeof(buf) - CLD_PKT_FTR_LEN, XDR_ENCODE);
+ if (!xdr_cld_pkt_hdr(&xdrs, &pkt)) {
+ HAIL_DEBUG(&sess->log, "%s: failed to encode header "
+ "for ack_seqid %lld", __func__, seqid_le);
+ xdr_destroy(&xdrs);
+ return -1009;
}
- return sess_send_pkt(sess, pkt, pkt_len);
+ /* Fill in footer */
+ hdr_len = xdr_getpos(&xdrs);
+ total_len = hdr_len + CLD_PKT_FTR_LEN;
+ foot = (struct cld_pkt_ftr *)(buf + hdr_len);
+ foot->seqid = seqid_le;
+ xdr_destroy(&xdrs);
+ ret = __cld_authsign(&sess->log, sess->user,
+ buf, total_len - SHA_DIGEST_LENGTH, foot->sha);
+ if (ret)
+ return ret;
+
+ return sess_send_pkt(sess, buf, total_len);
}
-static int cldc_rx_generic(struct cldc_session *sess,
- const struct cld_packet *pkt,
- const void *msgbuf,
- size_t buflen)
+static int rxmsg_generic(struct cldc_session *sess,
+ const struct cld_pkt_hdr *pkt,
+ const struct cld_pkt_ftr *foot)
{
- const struct cld_msg_resp *resp = msgbuf;
+ XDR xdrs;
+ struct cld_msg_generic_resp resp;
struct cldc_msg *req = NULL;
- ssize_t rc;
GList *tmp;
- if (buflen < sizeof(*resp))
+ xdrmem_create(&xdrs, sess->msg_buf, sess->msg_buf_len, XDR_DECODE);
+ if (!xdr_cld_msg_generic_resp(&xdrs, &resp)) {
+ HAIL_DEBUG(&sess->log, "%s: failed to decode "
+ "cld_msg_generic_resp", __func__);
+ xdr_destroy(&xdrs);
return -1008;
+ }
+ xdr_destroy(&xdrs);
+ /* Find out which outbound message this was a response to */
tmp = sess->out_msg;
while (tmp) {
req = tmp->data;
- HAIL_DEBUG(&sess->log, "rx_gen: comparing req->xid (%llu) "
- "with resp->xid_in (%llu)",
- (unsigned long long) le64_to_cpu(req->xid),
- (unsigned long long) le64_to_cpu(resp->xid_in));
+ HAIL_DEBUG(&sess->log, "%s: comparing req->xid (%llu) "
+ "with resp.xid_in (%llu)",
+ __func__,
+ (unsigned long long) req->xid,
+ (unsigned long long) resp.xid_in);
- if (req->xid == resp->xid_in)
+ if (req->xid == resp.xid_in)
break;
tmp = tmp->next;
}
- if (!tmp)
+ if (!tmp) {
+ HAIL_DEBUG(&sess->log, "%s: no match found with "
+ "xid_in %llu\n", __func__, resp.xid_in);
return -1005;
+ }
if (req->done) {
- HAIL_DEBUG(&sess->log, "rx_gen: re-acking");
+ HAIL_DEBUG(&sess->log, "%s: re-acking\n", __func__);
} else {
- HAIL_DEBUG(&sess->log, "rx_gen: issuing completion, acking");
-
+ HAIL_DEBUG(&sess->log, "%s: issuing completion, acking\n",
+ __func__);
req->done = true;
if (req->cb) {
- rc = req->cb(req, msgbuf, buflen, true);
+ ssize_t rc = req->cb(req, sess->msg_buf,
+ sess->msg_buf_len, resp.code);
if (rc < 0)
return rc;
}
}
- return ack_seqid(sess, pkt->seqid);
+ return ack_seqid(sess, foot->seqid);
}
-static int cldc_rx_ack_frag(struct cldc_session *sess,
- const struct cld_packet *pkt,
- const void *msgbuf,
- size_t buflen)
+static int rxmsg_ack_frag(struct cldc_session *sess,
+ const struct cld_pkt_hdr *pkt,
+ const struct cld_pkt_ftr *foot)
{
- const struct cld_msg_ack_frag *ack_msg = msgbuf;
- struct cldc_msg *req = NULL;
+ XDR xdrs;
+ struct cld_msg_ack_frag ack_msg;
GList *tmp;
- if (buflen < sizeof(*ack_msg))
+ xdrmem_create(&xdrs, sess->msg_buf, sess->msg_buf_len, XDR_DECODE);
+ memset(&ack_msg, 0, sizeof(ack_msg));
+ if (!xdr_cld_msg_ack_frag(&xdrs, &ack_msg)) {
+ HAIL_INFO(&sess->log, "%s: failed to decode ack_msg",
+ __func__);
+ xdr_destroy(&xdrs);
return -1008;
+ }
+ xdr_destroy(&xdrs);
- HAIL_DEBUG(&sess->log, "ack-frag: seqid %llu, want to ack",
- (unsigned long long) ack_msg->seqid);
+ HAIL_INFO(&sess->log, "ack-frag: seqid %llu, want to ack",
+ ack_msg.seqid);
tmp = sess->out_msg;
while (tmp) {
+ struct cldc_msg *req;
int i;
req = tmp->data;
@@ -188,15 +225,20 @@ static int cldc_rx_ack_frag(struct cldc_session *sess,
for (i = 0; i < req->n_pkts; i++) {
struct cldc_pkt_info *pi;
+ struct cld_pkt_ftr *f;
+ uint64_t seqid;
pi = req->pkt_info[i];
if (!pi)
continue;
- if (pi->pkt.seqid != ack_msg->seqid)
+ f = (struct cld_pkt_ftr *)
+ pi->data + (pi->pkt_len - CLD_PKT_FTR_LEN);
+ seqid = le64_to_cpu(f->seqid);
+ if (seqid != ack_msg.seqid)
continue;
HAIL_DEBUG(&sess->log, "ack-frag: seqid %llu, expiring",
- (unsigned long long) ack_msg->seqid);
+ (unsigned long long)ack_msg.seqid);
req->pkt_info[i] = NULL;
free(pi);
@@ -206,21 +248,27 @@ static int cldc_rx_ack_frag(struct cldc_session *sess,
return 0;
}
-static int cldc_rx_event(struct cldc_session *sess,
- const struct cld_packet *pkt,
- const void *msgbuf,
- size_t buflen)
+static int rxmsg_event(struct cldc_session *sess,
+ const struct cld_pkt_hdr *pkt,
+ const struct cld_pkt_ftr *foot)
{
- const struct cld_msg_event *ev = msgbuf;
+ XDR xdrs;
+ struct cld_msg_event ev;
struct cldc_fh *fh = NULL;
int i;
- if (buflen < sizeof(*ev))
+ xdrmem_create(&xdrs, sess->msg_buf, sess->msg_buf_len, XDR_DECODE);
+ if (!xdr_cld_msg_event(&xdrs, &ev)) {
+ HAIL_INFO(&sess->log, "%s: failed to decode cld_msg_event",
+ __func__);
+ xdr_destroy(&xdrs);
return -1008;
+ }
+ xdr_destroy(&xdrs);
for (i = 0; i < sess->fh->len; i++) {
fh = &g_array_index(sess->fh, struct cldc_fh, i);
- if (fh->fh_le == ev->fh)
+ if (fh->fh == ev.fh)
break;
else
fh = NULL;
@@ -229,21 +277,11 @@ static int cldc_rx_event(struct cldc_session *sess,
if (!fh)
return -1011;
- sess->ops->event(sess->private, sess, fh,
- le32_to_cpu(ev->events));
+ sess->ops->event(sess->private, sess, fh, ev.events);
return 0;
}
-static int cldc_rx_not_master(struct cldc_session *sess,
- const struct cld_packet *pkt,
- const void *msgbuf,
- size_t buflen)
-{
- HAIL_DEBUG(&sess->log, "FIXME: not-master message received");
- return -1055; /* FIXME */
-}
-
static void cldc_msg_free(struct cldc_msg *msg)
{
int i;
@@ -251,7 +289,7 @@ static void cldc_msg_free(struct cldc_msg *msg)
if (!msg)
return;
- for (i = 0; i < CLD_MAX_PKT_MSG; i++)
+ for (i = 0; i < msg->n_pkts; i++)
free(msg->pkt_info[i]);
free(msg);
@@ -278,214 +316,130 @@ static void sess_expire_outmsg(struct cldc_session *sess, time_t current_time)
sess->msg_scan_time = current_time + CLDC_MSG_SCAN;
}
-static const char *user_key(struct cldc_session *sess, const char *user)
+static int rx_complete(struct cldc_session *sess,
+ const struct cld_pkt_hdr *pkt,
+ const struct cld_pkt_ftr *foot)
{
- if (strcmp(sess->user, user))
- return NULL;
-
- return sess->secret_key;
-}
-
-static bool authcheck(struct cldc_session *sess, const struct cld_packet *pkt,
- size_t buflen)
-{
- size_t userlen = strnlen(pkt->user, sizeof(pkt->user));
- const char *key;
- unsigned char md[SHA_DIGEST_LENGTH];
- unsigned int md_len = 0;
- const void *buf = pkt;
-
- /* forbid zero-len and max-len (no nul) usernames */
- if (userlen < 1 || userlen >= sizeof(pkt->user))
- return false;
-
- key = user_key(sess, pkt->user);
- if (!key)
- return false;
-
- HMAC(EVP_sha1(), key, strlen(key), buf, buflen - SHA_DIGEST_LENGTH,
- md, &md_len);
-
- if (md_len != SHA_DIGEST_LENGTH)
- HAIL_INFO(&sess->log,
- "authsign BUG: md_len != SHA_DIGEST_LENGTH");
-
- if (memcmp(buf + buflen - SHA_DIGEST_LENGTH, md, SHA_DIGEST_LENGTH))
- return false;
-
- return true;
+ switch (sess->msg_buf_type) {
+ case CMT_ACK:
+ HAIL_INFO(&sess->log, "%s: received unexpected ACK\n",
+ __func__);
+ return -EBADRQC;
+ case CMT_PING:
+ /* send out an ACK */
+ return ack_seqid(sess, foot->seqid);
+ case CMT_NOT_MASTER:
+ HAIL_INFO(&sess->log, "FIXME: not-master message received");
+ return -1055; /* FIXME */
+ case CMT_EVENT:
+ return rxmsg_event(sess, pkt, foot);
+ case CMT_ACK_FRAG:
+ return rxmsg_ack_frag(sess, pkt, foot);
+ default:
+ return rxmsg_generic(sess, pkt, foot);
+ }
}
-static bool authsign(struct cldc_session *sess, struct cld_packet *pkt,
- size_t buflen)
+/** Accepts a packet's sequence ID.
+ * Depending on the message type, this may involve initializing the session's
+ * sequence ID, validating that the packet's ID is in range, or doing nothing.
+ *
+ * @param sess The session
+ * @param seqid The sequence ID
+ * @param type The message type
+ *
+ * @return 0 on success; error code otherwise
+ */
+static int accept_seqid(struct cldc_session *sess, uint64_t seqid,
+ enum cld_msg_type type)
{
- const char *key;
- unsigned char md[SHA_DIGEST_LENGTH];
- unsigned int md_len = 0;
- void *buf = pkt;
-
- key = user_key(sess, pkt->user);
- if (!key)
- return false;
-
- HMAC(EVP_sha1(), key, strlen(key), buf, buflen - SHA_DIGEST_LENGTH,
- md, &md_len);
-
- if (md_len != SHA_DIGEST_LENGTH)
- HAIL_INFO(&sess->log,
- "authsign BUG: md_len != SHA_DIGEST_LENGTH");
-
- memcpy(buf + (buflen - SHA_DIGEST_LENGTH), md, SHA_DIGEST_LENGTH);
-
- return true;
-}
+ switch (type) {
+ case CMT_NEW_SESS:
+ /* CMT_NEW_SESS initializes the session's sequence id */
+ sess->next_seqid_in = seqid + 1;
+ sess->next_seqid_in_tr =
+ sess->next_seqid_in - CLDC_MSG_REMEMBER;
+ HAIL_DEBUG(&sess->log, "%s: setting next_seqid_in to %llu",
+ __func__, (unsigned long long) seqid);
+ return 0;
-static const char *opstr(enum cld_msg_ops op)
-{
- switch (op) {
- case cmo_nop: return "cmo_nop";
- case cmo_new_sess: return "cmo_new_sess";
- case cmo_open: return "cmo_open";
- case cmo_get_meta: return "cmo_get_meta";
- case cmo_get: return "cmo_get";
- case cmo_put: return "cmo_put";
- case cmo_close: return "cmo_close";
- case cmo_del: return "cmo_del";
- case cmo_lock: return "cmo_lock";
- case cmo_unlock: return "cmo_unlock";
- case cmo_trylock: return "cmo_trylock";
- case cmo_ack: return "cmo_ack";
- case cmo_end_sess: return "cmo_end_sess";
- case cmo_ping: return "cmo_ping";
- case cmo_not_master: return "cmo_not_master";
- case cmo_event: return "cmo_event";
- case cmo_ack_frag: return "cmo_ack_frag";
- default: return "(unknown)";
- }
-}
+ case CMT_NOT_MASTER:
+ case CMT_ACK_FRAG:
+ /* Ignore sequence ID of these types */
+ return 0;
-static int cldc_receive_msg(struct cldc_session *sess,
- const struct cld_packet *pkt,
- size_t pkt_len)
-{
- const struct cld_msg_hdr *msg = (struct cld_msg_hdr *) sess->msg_buf;
- size_t msglen = sess->msg_buf_len;
+ default:
+ /* verify that the sequence id is in range */
+ if (seqid == sess->next_seqid_in) {
+ sess->next_seqid_in++;
+ sess->next_seqid_in_tr++;
+ return 0;
+ }
- if (memcmp(msg->magic, CLD_MSG_MAGIC, sizeof(msg->magic))) {
- HAIL_DEBUG(&sess->log, "receive_pkt: bad msg magic");
- return -EPROTO;
- }
+ if (seqid_in_range(seqid,
+ sess->next_seqid_in_tr,
+ sess->next_seqid_in)) {
+ return 0;
+ }
- switch(msg->op) {
- case cmo_nop:
- case cmo_close:
- case cmo_del:
- case cmo_lock:
- case cmo_unlock:
- case cmo_trylock:
- case cmo_put:
- case cmo_new_sess:
- case cmo_end_sess:
- case cmo_open:
- case cmo_get_meta:
- case cmo_get:
- return cldc_rx_generic(sess, pkt, msg, msglen);
- case cmo_not_master:
- return cldc_rx_not_master(sess, pkt, msg, msglen);
- case cmo_ack_frag:
- return cldc_rx_ack_frag(sess, pkt, msg, msglen);
- case cmo_event:
- return cldc_rx_event(sess, pkt, msg, msglen);
- case cmo_ping:
- return ack_seqid(sess, pkt->seqid);
- case cmo_ack:
- return -EBADRQC;
+ return -EBADSLT;
}
-
- /* unknown op code */
- return -EBADRQC;
}
int cldc_receive_pkt(struct cldc_session *sess,
const void *net_addr, size_t net_addrlen,
const void *pktbuf, size_t pkt_len)
{
- const struct cld_packet *pkt = pktbuf;
- const struct cld_msg_hdr *msg = (struct cld_msg_hdr *) (pkt + 1);
- size_t msglen;
+ int ret;
struct timeval tv;
time_t current_time;
+ struct cld_pkt_hdr pkt;
+ unsigned int hdr_len, msg_len;
+ const struct cld_pkt_ftr *foot;
uint64_t seqid;
- uint32_t pkt_flags;
- bool first_frag, last_frag, have_new_sess, no_seqid;
- bool have_get;
+ XDR xdrs;
gettimeofday(&tv, NULL);
current_time = tv.tv_sec;
- if (pkt_len < (sizeof(*pkt) + SHA_DIGEST_LENGTH)) {
- HAIL_DEBUG(&sess->log, "receive_pkt: msg too short");
+ /* Decode the packet header */
+ if (pkt_len < CLD_PKT_FTR_LEN) {
+ HAIL_DEBUG(&sess->log, "%s: packet too short to have a "
+ "well-formed footer\n", __func__);
return -EPROTO;
}
-
- msglen = pkt_len - sizeof(*pkt) - SHA_DIGEST_LENGTH;
-
- pkt_flags = le32_to_cpu(pkt->flags);
- first_frag = pkt_flags & CPF_FIRST;
- last_frag = pkt_flags & CPF_LAST;
- have_get = first_frag && (msg->op == cmo_get);
- have_new_sess = first_frag && (msg->op == cmo_new_sess);
- no_seqid = first_frag && ((msg->op == cmo_not_master) ||
- (msg->op == cmo_ack_frag));
-
- if (sess->log.verbose) {
- if (have_get) {
- struct cld_msg_get_resp *dp;
- dp = (struct cld_msg_get_resp *) msg;
- HAIL_DEBUG(&sess->log, "receive_pkt(len %u, op %s"
- ", seqid %llu, user %s, size %u)",
- (unsigned int) pkt_len,
- opstr(msg->op),
- (unsigned long long) le64_to_cpu(pkt->seqid),
- pkt->user,
- le32_to_cpu(dp->size));
- } else if (have_new_sess) {
- struct cld_msg_resp *dp;
- dp = (struct cld_msg_resp *) msg;
- HAIL_DEBUG(&sess->log, "receive_pkt(len %u, op %s"
- ", seqid %llu, user %s, xid_in %llu)",
- (unsigned int) pkt_len,
- opstr(msg->op),
- (unsigned long long) le64_to_cpu(pkt->seqid),
- pkt->user,
- (unsigned long long) le64_to_cpu(dp->xid_in));
- } else {
- HAIL_DEBUG(&sess->log, "receive_pkt(len %u, "
- "flags %s%s, op %s, seqid %llu, user %s)",
- (unsigned int) pkt_len,
- first_frag ? "F" : "",
- last_frag ? "L" : "",
- first_frag ? opstr(msg->op) : "n/a",
- (unsigned long long) le64_to_cpu(pkt->seqid),
- pkt->user);
- }
+ xdrmem_create(&xdrs, (void *)pktbuf,
+ pkt_len - CLD_PKT_FTR_LEN, XDR_DECODE);
+ memset(&pkt, 0, sizeof(pkt));
+ if (!xdr_cld_pkt_hdr(&xdrs, &pkt)) {
+ HAIL_DEBUG(&sess->log, "%s: failed to decode packet header\n",
+ __func__);
+ xdr_destroy(&xdrs);
+ return -EPROTO;
}
-
- if (memcmp(pkt->magic, CLD_PKT_MAGIC, sizeof(pkt->magic))) {
- HAIL_DEBUG(&sess->log, "receive_pkt: bad pkt magic");
+ hdr_len = xdr_getpos(&xdrs);
+ xdr_destroy(&xdrs);
+ if (memcmp(&pkt.magic, CLD_PKT_MAGIC, sizeof(pkt.magic))) {
+ HAIL_DEBUG(&sess->log, "%s: bad pkt magic\n", __func__);
return -EPROTO;
}
/* check HMAC signature */
- if (!authcheck(sess, pkt, pkt_len)) {
- HAIL_DEBUG(&sess->log, "receive_pkt: invalid auth");
+ foot = (const struct cld_pkt_ftr *)
+ (((char *)pktbuf) + (pkt_len - CLD_PKT_FTR_LEN));
+ ret = __cld_authcheck(&sess->log, pkt.user,
+ pktbuf, pkt_len - SHA_DIGEST_LENGTH, foot->sha);
+ if (ret) {
+ HAIL_DEBUG(&sess->log, "%s: invalid auth (ret=%d)\n",
+ __func__, ret);
return -EACCES;
}
/* verify stored server addr matches pkt addr */
if (((sess->addr_len != net_addrlen) ||
memcmp(sess->addr, net_addr, net_addrlen))) {
- HAIL_DEBUG(&sess->log, "receive_pkt: server address mismatch");
+ HAIL_DEBUG(&sess->log, "%s: server address mismatch",
+ __func__);
return -EBADE;
}
@@ -493,48 +447,39 @@ int cldc_receive_pkt(struct cldc_session *sess,
if (current_time >= sess->msg_scan_time)
sess_expire_outmsg(sess, current_time);
- if (first_frag)
- sess->msg_buf_len = 0;
-
- if ((sess->msg_buf_len + msglen) > CLD_MAX_MSG_SZ) {
- HAIL_DEBUG(&sess->log, "receive_pkt: bad pkt length");
- return -EPROTO;
+ if (pkt.mi.order & CLD_PKT_IS_FIRST) {
+ /* This packet begins a new message.
+ * Determine the new message's type */
+ sess->msg_buf_type = pkt.mi.cld_pkt_msg_info_u.mi.type;
}
- memcpy(sess->msg_buf + sess->msg_buf_len, msg, msglen);
- sess->msg_buf_len += msglen;
-
- /* verify (or set, for new-sess) sequence id */
- seqid = le64_to_cpu(pkt->seqid);
- if (have_new_sess) {
- sess->next_seqid_in = seqid + 1;
- sess->next_seqid_in_tr =
- sess->next_seqid_in - CLDC_MSG_REMEMBER;
-
- HAIL_DEBUG(&sess->log, "receive_pkt: "
- "setting next_seqid_in to %llu",
- (unsigned long long) seqid);
- } else if (!no_seqid) {
- if (seqid != sess->next_seqid_in) {
- if (seqid_in_range(seqid,
- sess->next_seqid_in_tr,
- sess->next_seqid_in))
- return ack_seqid(sess, pkt->seqid);
-
- HAIL_DEBUG(&sess->log, "receive_pkt: bad seqid %llu",
- (unsigned long long) seqid);
- return -EBADSLT;
- }
- sess->next_seqid_in++;
- sess->next_seqid_in_tr++;
+ seqid = le64_to_cpu(foot->seqid);
+ ret = accept_seqid(sess, seqid, sess->msg_buf_type);
+ if (ret) {
+ HAIL_DEBUG(&sess->log, "%s: bad seqid %llu",
+ __func__, (unsigned long long) seqid);
+ return ret;
}
+ if (pkt.mi.order & CLD_PKT_IS_FIRST)
+ sess->msg_buf_len = 0;
+ msg_len = pkt_len - hdr_len - CLD_PKT_FTR_LEN;
+ if ((sess->msg_buf_len + msg_len) > CLD_MAX_MSG_SZ) {
+ HAIL_DEBUG(&sess->log, "%s: message too long\n", __func__);
+ return -EPROTO;
+ }
+ memcpy(sess->msg_buf + sess->msg_buf_len, pktbuf + hdr_len, msg_len);
+ sess->msg_buf_len += msg_len;
sess->expire_time = current_time + CLDC_SESS_EXPIRE;
- if (!last_frag)
- return sess ? ack_seqid(sess, pkt->seqid) : 0;
-
- return cldc_receive_msg(sess, pkt, pkt_len);
+ if (pkt.mi.order & CLD_PKT_IS_LAST) {
+ HAIL_DEBUG(&sess->log, "%s: receiving complete message of "
+ "type %s\n", __func__,
+ __cld_msg_type_to_str(sess->msg_buf_type));
+ return rx_complete(sess, &pkt, foot);
+ } else {
+ return ack_seqid(sess, foot->seqid);
+ }
}
static void sess_next_seqid(struct cldc_session *sess, uint64_t *seqid)
@@ -543,68 +488,114 @@ static void sess_next_seqid(struct cldc_session *sess, uint64_t *seqid)
*seqid = rc;
}
+/**
+ * creates a new cldc_msg
+ *
+ * @param sess The session
+ * @param copts The call options
+ * @param type The type of message to create
+ * @param xdrproc The XDR function to use to create the message body
+ * @param data The data to pass to xdrproc
+ *
+ * @return The cldc message, or NULL on error,
+ */
static struct cldc_msg *cldc_new_msg(struct cldc_session *sess,
- const struct cldc_call_opts *copts,
- enum cld_msg_ops op,
- size_t msg_len)
+ const struct cldc_call_opts *copts,
+ enum cld_msg_type type,
+ xdrproc_t xdrproc, const void *data)
{
struct cldc_msg *msg;
- struct cld_msg_hdr *hdr;
struct timeval tv;
- int i, data_left;
- void *p;
+ size_t i, body_len, n_pkts;
+ char *body;
+ XDR xbdy;
+
+ /* Encode the message body */
+ body_len = xdr_sizeof(xdrproc, (void *)data);
+ body = alloca(body_len);
+ xdrmem_create(&xbdy, body, body_len, XDR_ENCODE);
+ if (!xdrproc(&xbdy, (void *)data)) {
+ HAIL_DEBUG(&sess->log, "%s: failed to encode "
+ "message", __func__);
+ xdr_destroy(&xbdy);
+ return NULL;
+ }
+ xdr_destroy(&xbdy);
- gettimeofday(&tv, NULL);
+ if (body_len == 0)
+ /* Some packets (like ACKS) just have a header, and no message
+ * body. */
+ n_pkts = 1;
+ else {
+ /* round up */
+ n_pkts = (body_len + CLD_MAX_PKT_MSG_SZ - 1) /
+ CLD_MAX_PKT_MSG_SZ;
+ }
- msg = calloc(1, sizeof(*msg) + msg_len);
+ /* Create cldc_msg */
+ msg = calloc(1, sizeof(struct cldc_msg) +
+ (sizeof(struct cldc_pkt_info *) * n_pkts));
if (!msg)
return NULL;
+ msg->n_pkts = n_pkts;
__cld_rand64(&msg->xid);
-
+ msg->type = type;
msg->sess = sess;
-
if (copts)
memcpy(&msg->copts, copts, sizeof(msg->copts));
-
+ gettimeofday(&tv, NULL);
msg->expire_time = tv.tv_sec + CLDC_MSG_EXPIRE;
- msg->data_len = msg_len;
-
- msg->n_pkts = msg_len / CLD_MAX_PKT_MSG_SZ;
- msg->n_pkts += ((msg_len % CLD_MAX_PKT_MSG_SZ) ? 1 : 0);
-
- p = msg->data;
- data_left = msg_len;
for (i = 0; i < msg->n_pkts; i++) {
+ XDR xhdr;
+ struct cld_pkt_hdr pkt;
struct cldc_pkt_info *pi;
- int pkt_len;
-
- pkt_len = MIN(data_left, CLD_MAX_PKT_MSG_SZ);
+ int hdr_len, body_chunk_len, pkt_len;
+
+ /* Set up packet header */
+ memcpy(&pkt.magic, CLD_PKT_MAGIC, sizeof(pkt.magic));
+ memcpy(&pkt.sid, sess->sid, CLD_SID_SZ);
+ pkt.user = sess->user;
+ if (i == 0) {
+ if (i == (msg->n_pkts - 1))
+ pkt.mi.order = CLD_PKT_ORD_FIRST_LAST;
+ else
+ pkt.mi.order = CLD_PKT_ORD_FIRST;
+ pkt.mi.cld_pkt_msg_info_u.mi.xid = msg->xid;
+ pkt.mi.cld_pkt_msg_info_u.mi.type = type;
+ } else {
+ if (i == (msg->n_pkts - 1))
+ pkt.mi.order = CLD_PKT_ORD_LAST;
+ else
+ pkt.mi.order = CLD_PKT_ORD_MID;
+ }
- pi = calloc(1, sizeof(*pi) + pkt_len + SHA_DIGEST_LENGTH);
+ /* Allocate memory */
+ hdr_len = xdr_sizeof((xdrproc_t)xdr_cld_pkt_hdr, &pkt);
+ body_chunk_len = MIN(body_len, CLD_MAX_PKT_MSG_SZ);
+ pkt_len = hdr_len + body_chunk_len + CLD_PKT_FTR_LEN;
+ pi = calloc(1, sizeof(*pi) + pkt_len);
if (!pi)
goto err_out;
-
pi->pkt_len = pkt_len;
-
- memcpy(pi->pkt.magic, CLD_PKT_MAGIC, CLD_MAGIC_SZ);
- memcpy(pi->pkt.sid, sess->sid, CLD_SID_SZ);
- strncpy(pi->pkt.user, sess->user, CLD_MAX_USERNAME - 1);
-
- if (i == 0)
- pi->pkt.flags |= cpu_to_le32(CPF_FIRST);
- if (i == (msg->n_pkts - 1))
- pi->pkt.flags |= cpu_to_le32(CPF_LAST);
-
msg->pkt_info[i] = pi;
- data_left -= pkt_len;
- }
+ strncpy(pi->user, sess->user, CLD_MAX_USERNAME - 1);
+
+ /* Fill in the packet header */
+ xdrmem_create(&xhdr, (char *)pi->data, hdr_len, XDR_ENCODE);
+ if (!xdr_cld_pkt_hdr(&xhdr, &pkt)) {
+ HAIL_DEBUG(&sess->log, "%s: failed to encode header "
+ "for packet %d", __func__, i);
+ xdr_destroy(&xhdr);
+ goto err_out;
+ }
- hdr = (struct cld_msg_hdr *) &msg->data[0];
- memcpy(&hdr->magic, CLD_MSG_MAGIC, CLD_MAGIC_SZ);
- hdr->op = op;
- hdr->xid = msg->xid;
+ /* Fill in the body */
+ memcpy(pi->data + hdr_len, body, body_chunk_len);
+ body += body_chunk_len;
+ body_len -= body_chunk_len;
+ }
return msg;
@@ -623,7 +614,7 @@ static void sess_msg_drop(struct cldc_session *sess)
tmp = tmp->next;
if (!msg->done && msg->cb)
- msg->cb(msg, NULL, 0, false);
+ msg->cb(msg, NULL, 0, CLE_TIMEOUT);
cldc_msg_free(msg);
}
@@ -644,31 +635,8 @@ static void sess_expire(struct cldc_session *sess)
}
static int sess_send_pkt(struct cldc_session *sess,
- const struct cld_packet *pkt, size_t pkt_len)
+ const void *pkt, size_t pkt_len)
{
- if (sess->log.verbose) {
- uint32_t flags = le32_to_cpu(pkt->flags);
- bool first = (flags & CPF_FIRST);
- bool last = (flags & CPF_LAST);
- uint8_t op = cmo_nop;
-
- if (first) {
- struct cld_msg_hdr *hdr;
-
- hdr = (struct cld_msg_hdr *) (pkt + 1);
- op = hdr->op;
- }
-
- HAIL_DEBUG(&sess->log,
- "send_pkt(len %zu, flags %s%s, "
- "op %s, seqid %llu)",
- pkt_len,
- first ? "F" : "",
- last ? "L" : "",
- first ? opstr(op) : "n/a",
- (unsigned long long) le64_to_cpu(pkt->seqid));
- }
-
return sess->ops->pkt_send(sess->private,
sess->addr, sess->addr_len,
pkt, pkt_len);
@@ -676,7 +644,6 @@ static int sess_send_pkt(struct cldc_session *sess,
static int sess_timer(struct cldc_session *sess, void *priv)
{
- struct cldc_msg *msg;
GList *tmp = sess->out_msg;
struct timeval tv;
@@ -687,28 +654,20 @@ static int sess_timer(struct cldc_session *sess, void *priv)
}
while (tmp) {
+ struct cldc_msg *msg = tmp->data;
int i;
- msg = tmp->data;
tmp = tmp->next;
if (msg->done)
continue;
for (i = 0; i < msg->n_pkts; i++) {
- struct cldc_pkt_info *pi;
- int total_pkt_len;
-
- pi = msg->pkt_info[i];
+ struct cldc_pkt_info *pi = msg->pkt_info[i];
if (!pi)
continue;
-
- total_pkt_len = sizeof(struct cld_packet) +
- pi->pkt_len + SHA_DIGEST_LENGTH;
-
pi->retries++;
-
- sess_send_pkt(sess, &pi->pkt, total_pkt_len);
+ sess_send_pkt(sess, pi->data, pi->pkt_len);
}
}
@@ -717,35 +676,30 @@ static int sess_timer(struct cldc_session *sess, void *priv)
return CLDC_MSG_RETRY;
}
-static int sess_send(struct cldc_session *sess,
- struct cldc_msg *msg)
+static int sess_send(struct cldc_session *sess, struct cldc_msg *msg)
{
- int i, data_left;
- void *p;
+ int ret, i;
- p = msg->data;
- data_left = msg->data_len;
for (i = 0; i < msg->n_pkts; i++) {
- struct cldc_pkt_info *pi;
- int total_pkt_len;
-
- pi = msg->pkt_info[i];
- memcpy(pi->data, p, pi->pkt_len);
-
- total_pkt_len = sizeof(struct cld_packet) +
- pi->pkt_len + SHA_DIGEST_LENGTH;
-
- sess_next_seqid(sess, &pi->pkt.seqid);
-
- p += pi->pkt_len;
- data_left -= pi->pkt_len;
-
- if (!authsign(sess, &pi->pkt, total_pkt_len))
- return -1;
+ struct cldc_pkt_info *pi = msg->pkt_info[i];
+ struct cld_pkt_ftr *foot;
+
+ /* Add the sequence number to the end of the packet */
+ foot = (struct cld_pkt_ftr *)
+ (pi->data + pi->pkt_len - CLD_PKT_FTR_LEN);
+ memset(foot, 0, CLD_PKT_FTR_LEN);
+ sess_next_seqid(sess, &foot->seqid);
+
+ /* Add the signature to the end of the packet */
+ ret = __cld_authsign(&sess->log, pi->user,
+ pi->data,
+ pi->pkt_len - SHA_DIGEST_LENGTH, foot->sha);
+ if (ret)
+ return ret;
/* attempt first send */
- if (sess_send_pkt(sess, &pi->pkt, total_pkt_len) < 0)
- return -1;
+ if (sess_send_pkt(sess, pi->data, pi->pkt_len) < 0)
+ return -EIO;
}
/* add to list of outgoing packets, waiting to be ack'd */
@@ -776,16 +730,8 @@ static void sess_free(struct cldc_session *sess)
}
static ssize_t end_sess_cb(struct cldc_msg *msg, const void *resp_p,
- size_t resp_len, bool ok)
+ size_t resp_len, enum cle_err_codes resp_rc)
{
- const struct cld_msg_resp *resp = resp_p;
- enum cle_err_codes resp_rc = CLE_OK;
-
- if (!ok)
- resp_rc = CLE_TIMEOUT;
- else
- resp_rc = le32_to_cpu(resp->code);
-
if (msg->copts.cb)
return msg->copts.cb(&msg->copts, resp_rc);
@@ -793,7 +739,8 @@ static ssize_t end_sess_cb(struct cldc_msg *msg, const void *resp_p,
return 0;
}
-int cldc_end_sess(struct cldc_session *sess, const struct cldc_call_opts *copts)
+int cldc_end_sess(struct cldc_session *sess,
+ const struct cldc_call_opts *copts)
{
struct cldc_msg *msg;
@@ -801,8 +748,8 @@ int cldc_end_sess(struct cldc_session *sess, const struct cldc_call_opts *copts)
return -EINVAL;
/* create END-SESS message */
- msg = cldc_new_msg(sess, copts, cmo_end_sess,
- sizeof(struct cld_msg_hdr));
+ msg = cldc_new_msg(sess, copts, CMT_END_SESS,
+ (xdrproc_t)xdr_void, NULL);
if (!msg)
return -ENOMEM;
@@ -812,19 +759,10 @@ int cldc_end_sess(struct cldc_session *sess, const struct cldc_call_opts *copts)
}
static ssize_t new_sess_cb(struct cldc_msg *msg, const void *resp_p,
- size_t resp_len, bool ok)
+ size_t resp_len, enum cle_err_codes resp_rc)
{
- struct cldc_session *sess = msg->sess;
- const struct cld_msg_resp *resp = resp_p;
- enum cle_err_codes resp_rc = CLE_OK;
-
- if (!ok)
- resp_rc = CLE_TIMEOUT;
- else
- resp_rc = le32_to_cpu(resp->code);
-
if (resp_rc == CLE_OK)
- sess->confirmed = true;
+ msg->sess->confirmed = true;
if (msg->copts.cb)
return msg->copts.cb(&msg->copts, resp_rc);
@@ -845,7 +783,6 @@ int cldc_new_sess(const struct cldc_ops *ops,
if (addr_len > sizeof(sess->addr))
return -EINVAL;
-
if (!user || !*user || !secret_key || !*secret_key)
return -EINVAL;
if (strlen(user) >= sizeof(sess->user))
@@ -877,8 +814,8 @@ int cldc_new_sess(const struct cldc_ops *ops,
sess->addr_len = addr_len;
/* create NEW-SESS message */
- msg = cldc_new_msg(sess, copts, cmo_new_sess,
- sizeof(struct cld_msg_hdr));
+ msg = cldc_new_msg(sess, copts, CMT_NEW_SESS,
+ (xdrproc_t)xdr_void, NULL);
if (!msg) {
sess_free(sess);
return -ENOMEM;
@@ -908,16 +845,8 @@ void cldc_kill_sess(struct cldc_session *sess)
}
static ssize_t generic_end_cb(struct cldc_msg *msg, const void *resp_p,
- size_t resp_len, bool ok)
+ size_t resp_len, enum cle_err_codes resp_rc)
{
- const struct cld_msg_resp *resp = resp_p;
- enum cle_err_codes resp_rc = CLE_OK;
-
- if (!ok)
- resp_rc = CLE_TIMEOUT;
- else
- resp_rc = le32_to_cpu(resp->code);
-
if (msg->copts.cb)
return msg->copts.cb(&msg->copts, resp_rc);
@@ -932,7 +861,8 @@ int cldc_nop(struct cldc_session *sess, const struct cldc_call_opts *copts)
return -EINVAL;
/* create NOP message */
- msg = cldc_new_msg(sess, copts, cmo_nop, sizeof(struct cld_msg_hdr));
+ msg = cldc_new_msg(sess, copts, CMT_NOP,
+ (xdrproc_t)xdr_void, NULL);
if (!msg)
return -ENOMEM;
@@ -945,8 +875,7 @@ int cldc_del(struct cldc_session *sess, const struct cldc_call_opts *copts,
const char *pathname)
{
struct cldc_msg *msg;
- struct cld_msg_del *del;
- void *p;
+ struct cld_msg_del del;
size_t plen;
if (!sess->confirmed)
@@ -957,46 +886,37 @@ int cldc_del(struct cldc_session *sess, const struct cldc_call_opts *copts,
return -EINVAL;
plen = strlen(pathname);
- if (plen > 65530)
+ if (plen > CLD_INODE_NAME_MAX)
return -EINVAL;
/* create DEL message */
- msg = cldc_new_msg(sess, copts, cmo_del,
- sizeof(struct cld_msg_del) + strlen(pathname));
+ del.inode_name = (char *)pathname;
+ msg = cldc_new_msg(sess, copts, CMT_DEL,
+ (xdrproc_t)xdr_cld_msg_del, &del);
if (!msg)
return -ENOMEM;
msg->cb = generic_end_cb;
- /* fill in DEL-specific name_len, name info */
- del = (struct cld_msg_del *) msg->data;
- del->name_len = cpu_to_le16(plen);
- p = del;
- p += sizeof(struct cld_msg_del);
- memcpy(p, pathname, plen);
-
return sess_send(sess, msg);
}
static ssize_t open_end_cb(struct cldc_msg *msg, const void *resp_p,
- size_t resp_len, bool ok)
+ size_t resp_len, enum cle_err_codes resp_rc)
{
- const struct cld_msg_open_resp *resp = resp_p;
- struct cldc_fh *fh = msg->cb_private;
- enum cle_err_codes resp_rc = CLE_OK;
-
- if (!ok)
- resp_rc = CLE_TIMEOUT;
- else {
- if (resp_len < sizeof(resp->resp))
+ if (resp_rc == CLE_OK) {
+ struct cldc_fh *fh = msg->cb_private;
+ XDR xdrs;
+ struct cld_msg_open_resp resp;
+
+ xdrmem_create(&xdrs, (void *)resp_p, resp_len, XDR_DECODE);
+ memset(&resp, 0, sizeof(resp));
+ if (!xdr_cld_msg_open_resp(&xdrs, &resp)) {
+ xdr_destroy(&xdrs);
return -1009;
- resp_rc = le32_to_cpu(resp->resp.code);
- }
+ }
- if (resp_rc == CLE_OK) {
- if (resp_len < sizeof(*resp))
- return -1010;
- fh->fh_le = resp->fh;
+ fh->fh = resp.fh;
fh->valid = true;
}
@@ -1012,9 +932,8 @@ int cldc_open(struct cldc_session *sess,
uint32_t events, struct cldc_fh **fh_out)
{
struct cldc_msg *msg;
- struct cld_msg_open *open;
+ struct cld_msg_open open;
struct cldc_fh fh, *fhtmp;
- void *p;
size_t plen;
int fh_idx;
@@ -1028,12 +947,15 @@ int cldc_open(struct cldc_session *sess,
return -EINVAL;
plen = strlen(pathname);
- if (plen > 65530)
+ if (plen > CLD_INODE_NAME_MAX)
return -EINVAL;
/* create OPEN message */
- msg = cldc_new_msg(sess, copts, cmo_open,
- sizeof(struct cld_msg_open) + strlen(pathname));
+ open.mode = open_mode;
+ open.events = events;
+ open.inode_name = (char *)pathname;
+ msg = cldc_new_msg(sess, copts, CMT_OPEN,
+ (xdrproc_t)xdr_cld_msg_open, &open);
if (!msg)
return -ENOMEM;
@@ -1048,15 +970,6 @@ int cldc_open(struct cldc_session *sess,
msg->cb = open_end_cb;
msg->cb_private = fhtmp;
- /* fill in OPEN-specific info */
- open = (struct cld_msg_open *) msg->data;
- open->mode = cpu_to_le32(open_mode);
- open->events = cpu_to_le32(events);
- open->name_len = cpu_to_le16(plen);
- p = open;
- p += sizeof(struct cld_msg_open);
- memcpy(p, pathname, plen);
-
*fh_out = fhtmp;
return sess_send(sess, msg);
@@ -1066,7 +979,7 @@ int cldc_close(struct cldc_fh *fh, const struct cldc_call_opts *copts)
{
struct cldc_session *sess;
struct cldc_msg *msg;
- struct cld_msg_close *close_msg;
+ struct cld_msg_close close;
if (!fh->valid)
return -EINVAL;
@@ -1074,8 +987,9 @@ int cldc_close(struct cldc_fh *fh, const struct cldc_call_opts *copts)
sess = fh->sess;
/* create CLOSE message */
- msg = cldc_new_msg(sess, copts, cmo_close,
- sizeof(struct cld_msg_close));
+ close.fh = fh->fh;
+ msg = cldc_new_msg(sess, copts, CMT_CLOSE,
+ (xdrproc_t)xdr_cld_msg_close, &close);
if (!msg)
return -ENOMEM;
@@ -1084,10 +998,6 @@ int cldc_close(struct cldc_fh *fh, const struct cldc_call_opts *copts)
msg->cb = generic_end_cb;
- /* fill in CLOSE-specific fh info */
- close_msg = (struct cld_msg_close *) msg->data;
- close_msg->fh = fh->fh_le;
-
return sess_send(sess, msg);
}
@@ -1096,7 +1006,7 @@ int cldc_lock(struct cldc_fh *fh, const struct cldc_call_opts *copts,
{
struct cldc_session *sess;
struct cldc_msg *msg;
- struct cld_msg_lock *lock;
+ struct cld_msg_lock lock;
if (!fh->valid)
return -EINVAL;
@@ -1104,19 +1014,16 @@ int cldc_lock(struct cldc_fh *fh, const struct cldc_call_opts *copts,
sess = fh->sess;
/* create LOCK message */
+ lock.fh = fh->fh;
+ lock.flags = lock_flags;
msg = cldc_new_msg(sess, copts,
- wait_for_lock ? cmo_lock : cmo_trylock,
- sizeof(struct cld_msg_lock));
+ wait_for_lock ? CMT_LOCK : CMT_TRYLOCK,
+ (xdrproc_t)xdr_cld_msg_lock, &lock);
if (!msg)
return -ENOMEM;
msg->cb = generic_end_cb;
- /* fill in LOCK-specific info */
- lock = (struct cld_msg_lock *) msg->data;
- lock->fh = fh->fh_le;
- lock->flags = cpu_to_le32(lock_flags);
-
return sess_send(sess, msg);
}
@@ -1124,7 +1031,7 @@ int cldc_unlock(struct cldc_fh *fh, const struct cldc_call_opts *copts)
{
struct cldc_session *sess;
struct cldc_msg *msg;
- struct cld_msg_unlock *unlock;
+ struct cld_msg_unlock unlock;
if (!fh->valid)
return -EINVAL;
@@ -1132,17 +1039,14 @@ int cldc_unlock(struct cldc_fh *fh, const struct cldc_call_opts *copts)
sess = fh->sess;
/* create UNLOCK message */
- msg = cldc_new_msg(sess, copts, cmo_unlock,
- sizeof(struct cld_msg_unlock));
+ unlock.fh = fh->fh;
+ msg = cldc_new_msg(sess, copts, CMT_UNLOCK,
+ (xdrproc_t)xdr_cld_msg_unlock, &unlock);
if (!msg)
return -ENOMEM;
msg->cb = generic_end_cb;
- /* fill in UNLOCK-specific info */
- unlock = (struct cld_msg_unlock *) msg->data;
- unlock->fh = fh->fh_le;
-
return sess_send(sess, msg);
}
@@ -1151,9 +1055,9 @@ int cldc_put(struct cldc_fh *fh, const struct cldc_call_opts *copts,
{
struct cldc_session *sess;
struct cldc_msg *msg;
- struct cld_msg_put *put;
+ struct cld_msg_put put;
- if (!data || !data_len || data_len > CLD_MAX_MSG_SZ)
+ if (!data || !data_len || data_len > CLD_MAX_PAYLOAD_SZ)
return -EINVAL;
if (!fh->valid)
@@ -1162,17 +1066,14 @@ int cldc_put(struct cldc_fh *fh, const struct cldc_call_opts *copts,
sess = fh->sess;
/* create PUT message */
- msg = cldc_new_msg(sess, copts, cmo_put,
- sizeof(struct cld_msg_put) + data_len);
+ put.fh = fh->fh;
+ put.data.data_len = data_len;
+ put.data.data_val = (char *)data;
+ msg = cldc_new_msg(sess, copts, CMT_PUT,
+ (xdrproc_t)xdr_cld_msg_put, &put);
if (!msg)
return -ENOMEM;
- put = (struct cld_msg_put *) msg->data;
- put->fh = fh->fh_le;
- put->data_size = cpu_to_le32(data_len);
-
- memcpy((put + 1), data, data_len);
-
msg->cb = generic_end_cb;
sess_send(sess, msg);
@@ -1180,73 +1081,39 @@ int cldc_put(struct cldc_fh *fh, const struct cldc_call_opts *copts,
return 0;
}
-#undef XC32
-#undef XC64
-#define XC32(name) \
- o->name = le32_to_cpu(resp->name)
-#define XC64(name) \
- o->name = le64_to_cpu(resp->name)
-
static ssize_t get_end_cb(struct cldc_msg *msg, const void *resp_p,
- size_t resp_len, bool ok)
+ size_t resp_len, enum cle_err_codes resp_rc)
{
- const struct cld_msg_get_resp *resp = resp_p;
- enum cle_err_codes resp_rc = CLE_OK;
- struct cld_msg_get_resp *o = NULL;
-
- if (!ok)
- resp_rc = CLE_TIMEOUT;
- else
- resp_rc = le32_to_cpu(resp->resp.code);
-
if (resp_rc == CLE_OK) {
- bool get_body;
-
- o = &msg->copts.u.get.resp;
-
- get_body = (resp->resp.hdr.op == cmo_get);
- msg->copts.op = cmo_get;
-
- /* copy-and-swap */
- XC64(inum);
- XC32(ino_len);
- XC32(size);
- XC64(version);
- XC64(time_create);
- XC64(time_modify);
- XC32(flags);
-
- /* copy inode name */
- if (o->ino_len <= CLD_INODE_NAME_MAX) {
- size_t diffsz;
- const void *p;
-
- p = (resp + 1);
- memcpy(&msg->copts.u.get.inode_name, p, o->ino_len);
-
- p += o->ino_len;
- diffsz = p - resp_p;
-
- /* point to internal buffer holding GET data */
- msg->copts.u.get.buf = msg->sess->msg_buf + diffsz;
- msg->copts.u.get.size = msg->sess->msg_buf_len - diffsz;
- } else {
- o->ino_len = 0; /* Probably full of garbage */
+ XDR xin;
+ struct cld_msg_get_resp *resp = &msg->copts.resp;
+
+ /* Parse GET response.
+ * Avoid memory allocation in xdr_string by pointing
+ * variable-length elements at static buffers. */
+ xdrmem_create(&xin, (void *)resp_p, resp_len, XDR_DECODE);
+ memset(resp, 0, sizeof(struct cld_msg_get_resp));
+ resp->inode_name = msg->sess->inode_name_temp;
+ resp->data.data_val = msg->sess->payload;
+ resp->data.data_len = 0;
+ if (!xdr_cld_msg_get_resp(&xin, resp)) {
+ xdr_destroy(&xin);
+ return -1009;
}
+ xdr_destroy(&xin);
}
if (msg->copts.cb)
return msg->copts.cb(&msg->copts, resp_rc);
return 0;
}
-#undef XC
int cldc_get(struct cldc_fh *fh, const struct cldc_call_opts *copts,
bool metadata_only)
{
struct cldc_session *sess;
struct cldc_msg *msg;
- struct cld_msg_get *get;
+ struct cld_msg_get get;
if (!fh->valid)
return -EINVAL;
@@ -1254,17 +1121,14 @@ int cldc_get(struct cldc_fh *fh, const struct cldc_call_opts *copts,
sess = fh->sess;
/* create GET message */
- msg = cldc_new_msg(sess, copts, cmo_get,
- sizeof(struct cld_msg_get));
+ get.fh = fh->fh;
+ msg = cldc_new_msg(sess, copts, CMT_GET,
+ (xdrproc_t)xdr_cld_msg_get, &get);
if (!msg)
return -ENOMEM;
msg->cb = get_end_cb;
- /* fill in GET-specific info */
- get = (struct cld_msg_get *) msg->data;
- get->fh = fh->fh_le;
-
return sess_send(sess, msg);
}
diff --git a/lib/common.c b/lib/common.c
index 276c805..27a3f4e 100644
--- a/lib/common.c
+++ b/lib/common.c
@@ -26,7 +26,11 @@
#include <errno.h>
#include <glib.h>
#include <cld-private.h>
-#include "cld_msg.h"
+#include <openssl/sha.h>
+#include <openssl/hmac.h>
+#include "cld_msg_rpc.h"
+#include <hail_log.h>
+#include <syslog.h>
/* duplicated from tools/cldcli.c; put in common header somewhere? */
#define ARRAY_SIZE(arr) (sizeof(arr) / sizeof((arr)[0]))
diff --git a/server/Makefile.am b/server/Makefile.am
index faaf099..17771de 100644
--- a/server/Makefile.am
+++ b/server/Makefile.am
@@ -1,13 +1,18 @@
-INCLUDES = -I$(top_srcdir)/include @GLIB_CFLAGS@ \
+INCLUDES = -I$(top_srcdir)/include \
+ -I$(top_srcdir)/lib \
+ @GLIB_CFLAGS@ \
-DCLD_LIBDIR=\""$(libdir)"\" \
-DCLD_LOCAL_STATE_DIR="\"$(localstatedir)\""
sbin_PROGRAMS = cld cldbadm
cld_SOURCES = cldb.h cld.h \
+ ../lib/cld_fmt.c \
../lib/common.c \
- cldb.c msg.c server.c session.c util.c
+ cldb.c msg.c server.c session.c util.c \
+ ../lib/cld_msg_rpc_xdr.c
+
cld_LDADD = ../lib/libtimer.a \
@CRYPTO_LIBS@ @GLIB_LIBS@ @DB4_LIBS@
diff --git a/server/cld.h b/server/cld.h
index e6bed3c..26121b8 100644
--- a/server/cld.h
+++ b/server/cld.h
@@ -25,7 +25,7 @@
#include <poll.h>
#include <glib.h>
#include "cldb.h"
-#include <cld_msg.h>
+#include <lib/cld_msg_rpc.h>
#include <cld_common.h>
#include <libtimer.h>
#include <hail_log.h>
@@ -73,20 +73,12 @@ struct session {
bool dead; /* session has ended */
/* huge buffer should always come last */
+ enum cld_msg_type msg_type;
+ uint64_t msg_xid;
unsigned int msg_buf_len;
char msg_buf[CLD_MAX_MSG_SZ];
};
-struct msg_params {
- int sock_fd;
- const struct client *cli;
- struct session *sess;
-
- const struct cld_packet *pkt;
- const void *msg;
- size_t msg_len;
-};
-
struct server_stats {
unsigned long poll; /* num. polls */
unsigned long event; /* events dispatched */
@@ -123,46 +115,89 @@ struct server {
struct server_stats stats; /* global statistics */
};
+struct pkt_info {
+ struct cld_pkt_hdr *pkt;
+ struct session *sess;
+ uint64_t seqid;
+ uint64_t xid;
+ enum cld_msg_type type;
+ size_t hdr_len;
+};
+
/* msg.c */
extern int inode_lock_rescan(DB_TXN *txn, cldino_t inum);
-extern void msg_open(struct msg_params *);
-extern void msg_put(struct msg_params *);
-extern void msg_close(struct msg_params *);
-extern void msg_del(struct msg_params *);
-extern void msg_unlock(struct msg_params *);
-extern void msg_lock(struct msg_params *, bool);
-extern void msg_ack(struct msg_params *);
-extern void msg_get(struct msg_params *, bool);
+extern void msg_get(struct session *sess, const void *v);
+extern void msg_open(struct session *sess, const void *v);
+extern void msg_put(struct session *sess, const void *v);
+extern void msg_close(struct session *sess, const void *v);
+extern void msg_del(struct session *sess, const void *v);
+extern void msg_unlock(struct session *sess, const void *v);
+extern void msg_lock(struct session *sess, const void *v);
+extern void msg_ack(struct session *sess, uint64_t seqid);
/* session.c */
extern uint64_t next_seqid_le(uint64_t *seq);
-extern void pkt_init_pkt(struct cld_packet *dest, const struct cld_packet *src);
extern guint sess_hash(gconstpointer v);
extern gboolean sess_equal(gconstpointer _a, gconstpointer _b);
-extern void msg_new_sess(struct msg_params *, const struct client *);
-extern void msg_end_sess(struct msg_params *, const struct client *);
+extern void msg_new_sess(int sock_fd, const struct client *cli,
+ const struct pkt_info *info);
+extern void msg_end_sess(struct session *sess, uint64_t xid);
extern struct raw_session *session_new_raw(const struct session *sess);
extern void sessions_free(void);
-extern bool sess_sendmsg(struct session *sess, const void *msg_, size_t msglen,
- void (*done_cb)(struct session_outpkt *),
- void *done_data);
+
+/** Send a message as part of a session.
+ *
+ * @param sess The session
+ * @param xdrproc The XDR function to use to serialize the data
+ * @param xdrdata The message data
+ * @param type The type of the message
+ * @param done_cb The callback to call when the message has been acked
+ * @param done_data The data to give to done_cb
+ *
+ * @return true only if the message was sent
+ */
+extern bool sess_sendmsg(struct session *sess,
+ xdrproc_t xdrproc, const void *xdrdata, enum cld_msg_type type,
+ void (*done_cb)(struct session_outpkt *), void *done_data);
+
+/** Send a generic response message.
+ *
+ * @param sess The session
+ * @param code The error code to send
+ */
+extern void sess_sendresp_generic(struct session *sess,
+ enum cle_err_codes code);
+
extern int session_dispose(DB_TXN *txn, struct session *sess);
extern int session_remove_locks(DB_TXN *txn, uint8_t *sid, uint64_t fh,
cldino_t inum, bool *waiter);
extern int sess_load(GHashTable *ss);
/* server.c */
-extern const char *opstr(enum cld_msg_ops op);
extern struct server cld_srv;
extern struct hail_log srv_log;
extern struct timeval current_time;
extern int udp_tx(int sock_fd, struct sockaddr *, socklen_t,
const void *, size_t);
-extern void resp_copy(struct cld_msg_resp *resp, const struct cld_msg_hdr *src);
-extern void resp_err(struct session *sess,
- const struct cld_msg_hdr *src, enum cle_err_codes errcode);
-extern void resp_ok(struct session *sess, const struct cld_msg_hdr *src);
-extern bool authsign(struct cld_packet *pkt, size_t pkt_len);
+
+/** Transmit a single packet.
+ *
+ * This function doesn't provide error-retransmission logic.
+ * It can't handle messages that are bigger than a single packet.
+ *
+ * @param fd Socket to send the response on
+ * @param cli Client address data
+ * @param sid The session-id to use. Must be of length CLD_SID_SZ
+ * @param seqid The sequence id to use
+ * @param xdrproc The XDR function to use to serialize the data
+ * @param xdrdata The message data
+ * @param type The type of message to send
+ *
+ * @return true only on success
+ */
+extern void simple_sendmsg(int fd, const struct client *cli,
+ uint64_t sid, const char *username, uint64_t seqid,
+ xdrproc_t xdrproc, const void *xdrdata, enum cld_msg_type type);
/* util.c */
extern int write_pid_file(const char *pid_fn);
diff --git a/server/cldb.h b/server/cldb.h
index ac108ce..71dce95 100644
--- a/server/cldb.h
+++ b/server/cldb.h
@@ -23,7 +23,7 @@
#include <stdbool.h>
#include <db.h>
#include <cld-private.h>
-#include <cld_msg.h>
+#include <lib/cld_msg_rpc.h>
typedef uint64_t cldino_t;
diff --git a/server/msg.c b/server/msg.c
index e58059e..60403d0 100644
--- a/server/msg.c
+++ b/server/msg.c
@@ -25,6 +25,8 @@
#include <syslog.h>
#include <openssl/sha.h>
#include <cld-private.h>
+#include <cld_common.h>
+#include <lib/cld_msg_rpc.h>
#include "cld.h"
enum {
@@ -194,7 +196,6 @@ static int inode_notify(DB_TXN *txn, cldino_t inum, bool deleted)
DB *hand_idx = cld_srv.cldb.handle_idx;
DBC *cur;
DBT key, val;
- struct cld_msg_event me;
cldino_t inum_le = cldino_to_le(inum);
int gflags;
struct session *sess;
@@ -210,10 +211,6 @@ static int inode_notify(DB_TXN *txn, cldino_t inum, bool deleted)
val.ulen = sizeof(h);
val.flags = DB_DBT_USERMEM;
- memset(&me, 0, sizeof(me));
- memcpy(me.hdr.magic, CLD_MSG_MAGIC, CLD_MAGIC_SZ);
- me.hdr.op = cmo_event;
-
rc = hand_idx->cursor(hand_idx, txn, &cur, 0);
if (rc) {
hand_idx->err(hand_idx, rc, "inode_notify cursor");
@@ -222,6 +219,8 @@ static int inode_notify(DB_TXN *txn, cldino_t inum, bool deleted)
gflags = DB_SET;
while (1) {
+ struct cld_msg_event me;
+
rc = cur->get(cur, &key, &val, gflags);
if (rc) {
if (rc != DB_NOTFOUND)
@@ -248,10 +247,12 @@ static int inode_notify(DB_TXN *txn, cldino_t inum, bool deleted)
continue;
}
- me.fh = h.fh;
- me.events = cpu_to_le32(deleted ? CE_DELETED : CE_UPDATED);
+ memset(&me, 0, sizeof(me));
+ me.fh = le64_to_cpu(h.fh);
+ me.events = deleted ? CE_DELETED : CE_UPDATED;
- if (!sess_sendmsg(sess, &me, sizeof(me), NULL, NULL))
+ if (!sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_event,
+ (void *)&me, CMT_EVENT, NULL, NULL))
break;
}
@@ -292,7 +293,6 @@ int inode_lock_rescan(DB_TXN *txn, cldino_t inum)
cldino_t inum_le = cldino_to_le(inum);
struct raw_lock lock;
uint32_t lflags;
- struct cld_msg_event me;
struct session *sess;
rc = db_locks->cursor(db_locks, txn, &cur, 0);
@@ -312,15 +312,13 @@ int inode_lock_rescan(DB_TXN *txn, cldino_t inum)
val.ulen = sizeof(lock);
val.flags = DB_DBT_USERMEM;
- memset(&me, 0, sizeof(me));
- memcpy(me.hdr.magic, CLD_MSG_MAGIC, CLD_MAGIC_SZ);
- me.hdr.op = cmo_event;
-
/* loop through locks associated with this inode, searching
* for pending locks that can be converted into acquired
*/
gflags = DB_SET | DB_RMW;
while (1) {
+ struct cld_msg_event me;
+
rc = cur->get(cur, &key, &val, gflags);
if (rc) {
/* no locks, or no next-dup */
@@ -376,10 +374,12 @@ int inode_lock_rescan(DB_TXN *txn, cldino_t inum)
continue;
}
- me.fh = lock.fh;
- me.events = cpu_to_le32(CE_LOCKED);
+ memset(&me, 0, sizeof(me));
+ me.fh = le64_to_cpu(lock.fh);
+ me.events = CE_LOCKED;
- if (!sess_sendmsg(sess, &me, sizeof(me), NULL, NULL))
+ if (!sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_event,
+ (void *)&me, CMT_EVENT, NULL, NULL))
break;
}
@@ -387,12 +387,10 @@ int inode_lock_rescan(DB_TXN *txn, cldino_t inum)
return rc;
}
-void msg_get(struct msg_params *mp, bool metadata_only)
+void msg_get(struct session *sess, const void *v)
{
- const struct cld_msg_get *msg = mp->msg;
- struct cld_msg_get_resp *resp;
- size_t resp_len;
- uint64_t fh;
+ const struct cld_msg_get *get = v;
+ struct cld_msg_get_resp resp;
struct raw_handle *h = NULL;
struct raw_inode *inode = NULL;
enum cle_err_codes resp_rc = CLE_OK;
@@ -400,17 +398,10 @@ void msg_get(struct msg_params *mp, bool metadata_only)
uint32_t name_len, inode_size;
uint32_t omode;
int rc;
- struct session *sess = mp->sess;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- void *p;
-
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- /* get filehandle from input msg */
- fh = le64_to_cpu(msg->fh);
+ void *data_mem;
+ char *inode_name;
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
@@ -420,7 +411,7 @@ void msg_get(struct msg_params *mp, bool metadata_only)
}
/* read handle from db */
- rc = cldb_handle_get(txn, sess->sid, fh, &h, 0);
+ rc = cldb_handle_get(txn, sess->sid, get->fh, &h, 0);
if (rc) {
resp_rc = CLE_FH_INVAL;
goto err_out;
@@ -441,41 +432,30 @@ void msg_get(struct msg_params *mp, bool metadata_only)
goto err_out;
}
- name_len = le32_to_cpu(inode->ino_len);
inode_size = le32_to_cpu(inode->size);
-
- resp_len = sizeof(*resp) + name_len +
- (metadata_only ? 0 : inode_size);
- resp = alloca(resp_len);
- if (!resp) {
- resp_rc = CLE_OOM;
- goto err_out;
- }
-
- HAIL_DEBUG(&srv_log, "GET-DEBUG: sizeof(resp) %zu, name_len %u, "
- "inode->size %u, resp_len %zu",
- sizeof(*resp), name_len,
- inode_size, resp_len);
+ HAIL_DEBUG(&srv_log, "GET-DEBUG: inode->size %u\n", inode_size);
/* return response containing inode metadata */
- memset(resp, 0, resp_len);
- resp_copy(&resp->resp, mp->msg);
- resp->inum = inode->inum;
- resp->ino_len = inode->ino_len;
- resp->size = inode->size;
- resp->version = inode->version;
- resp->time_create = inode->time_create;
- resp->time_modify = inode->time_modify;
- resp->flags = inode->flags;
+ memset(&resp, 0, sizeof(resp));
+ resp.msg.code = CLE_OK;
+ resp.msg.xid_in = sess->msg_xid;
+ resp.inum = le64_to_cpu(inode->inum);
+ resp.vers = le64_to_cpu(inode->version);
+ resp.time_create = le64_to_cpu(inode->time_create);
+ resp.time_modify = le64_to_cpu(inode->time_modify);
+ resp.flags = le32_to_cpu(inode->flags);
- p = (resp + 1);
- memcpy(p, (inode + 1), name_len);
+ name_len = le32_to_cpu(inode->ino_len);
+ inode_name = alloca(name_len + 1);
+ snprintf(inode_name, name_len + 1, "%s", (char *)(inode + 1));
+ resp.inode_name = inode_name;
- p += name_len;
+ resp.data.data_len = 0;
+ resp.data.data_val = NULL;
/* send data, if requested */
- if (!metadata_only) {
- void *data_mem;
+ data_mem = NULL;
+ if (sess->msg_type == CMT_GET) {
size_t data_mem_len;
rc = cldb_data_get(txn, inum, &data_mem, &data_mem_len,
@@ -484,22 +464,23 @@ void msg_get(struct msg_params *mp, bool metadata_only)
/* treat not-found as zero length file, as we may
* not yet have created the data record
*/
- if (rc == DB_NOTFOUND) {
- resp->size = 0;
- resp_len -= inode_size;
- } else if (rc || (data_mem_len != inode_size)) {
- if (!rc)
- free(data_mem);
- resp_rc = CLE_DB_ERR;
- goto err_out;
- } else {
- memcpy(p, data_mem, data_mem_len);
-
- free(data_mem);
+ if (rc != DB_NOTFOUND) {
+ if (rc || (data_mem_len != inode_size)) {
+ if (!rc)
+ free(data_mem);
+ resp_rc = CLE_DB_ERR;
+ goto err_out;
+ } else {
+ resp.data.data_len = data_mem_len;
+ resp.data.data_val = data_mem;
+ }
}
}
- sess_sendmsg(sess, resp, resp_len, NULL, NULL);
+ sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_get_resp,
+ (void *)&resp, CMT_GET, NULL, NULL);
+ if (data_mem)
+ free(data_mem);
rc = txn->commit(txn, 0);
if (rc)
@@ -514,14 +495,14 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_get txn abort");
err_out_noabort:
- resp_err(sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(h);
free(inode);
}
-void msg_open(struct msg_params *mp)
+void msg_open(struct session *sess, const void *v)
{
- const struct cld_msg_open *msg = mp->msg;
+ const struct cld_msg_open *open = v;
struct cld_msg_open_resp resp;
const char *name;
struct raw_session *raw_sess = NULL;
@@ -532,24 +513,12 @@ void msg_open(struct msg_params *mp)
struct pathname_info pinfo;
void *parent_data = NULL;
size_t parent_len;
- uint32_t msg_mode, msg_events;
uint64_t fh;
cldino_t inum;
enum cle_err_codes resp_rc = CLE_OK;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- msg_mode = le32_to_cpu(msg->mode);
- msg_events = le32_to_cpu(msg->events);
- name_len = le16_to_cpu(msg->name_len);
-
- if (mp->msg_len < (sizeof(*msg) + name_len))
- return;
-
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
dbenv->err(dbenv, rc, "DB_ENV->txn_begin");
@@ -557,11 +526,12 @@ void msg_open(struct msg_params *mp)
goto err_out_noabort;
}
- name = mp->msg + sizeof(*msg);
+ name = open->inode_name;
+ name_len = strlen(name);
- create = msg_mode & COM_CREATE;
- excl = msg_mode & COM_EXCL;
- do_dir = msg_mode & COM_DIRECTORY;
+ create = open->mode & COM_CREATE;
+ excl = open->mode & COM_EXCL;
+ do_dir = open->mode & COM_DIRECTORY;
if (!valid_inode_name(name, name_len) || (create && name_len < 2)) {
resp_rc = CLE_NAME_INVAL;
@@ -659,7 +629,7 @@ void msg_open(struct msg_params *mp)
inum = cldino_from_le(inode->inum);
/* alloc & init new handle; updates session's next_fh */
- h = cldb_handle_new(mp->sess, inum, msg_mode, msg_events);
+ h = cldb_handle_new(sess, inum, open->mode, open->events);
if (!h) {
HAIL_CRIT(&srv_log, "cannot allocate handle");
resp_rc = CLE_OOM;
@@ -686,7 +656,7 @@ void msg_open(struct msg_params *mp)
}
/* encode in-memory session to raw database session struct */
- raw_sess = session_new_raw(mp->sess);
+ raw_sess = session_new_raw(sess);
if (!raw_sess) {
HAIL_CRIT(&srv_log, "cannot allocate session");
@@ -714,11 +684,12 @@ void msg_open(struct msg_params *mp)
free(raw_sess);
free(h);
- resp_copy(&resp.resp, mp->msg);
- resp.resp.code = cpu_to_le32(CLE_OK);
- resp.fh = cpu_to_le64(fh);
- sess_sendmsg(mp->sess, &resp, sizeof(resp), NULL, NULL);
-
+ memset(&resp, 0, sizeof(resp));
+ resp.msg.xid_in = sess->msg_xid;
+ resp.msg.code = CLE_OK;
+ resp.fh = fh;
+ sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_open_resp,
+ (void *)&resp, CMT_OPEN, NULL, NULL);
return;
err_out:
@@ -726,7 +697,7 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_open txn abort");
err_out_noabort:
- resp_err(mp->sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(parent_data);
free(parent);
free(inode);
@@ -734,38 +705,18 @@ err_out_noabort:
free(h);
}
-void msg_put(struct msg_params *mp)
+void msg_put(struct session *sess, const void *v)
{
- const struct cld_msg_put *msg = mp->msg;
- struct session *sess = mp->sess;
- uint64_t fh;
+ const struct cld_msg_put *put = v;
struct raw_handle *h = NULL;
struct raw_inode *inode = NULL;
enum cle_err_codes resp_rc = CLE_OK;
- const void *mem;
int rc;
cldino_t inum;
- uint32_t omode, data_size;
+ uint32_t omode;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- /* make sure input data as large as message header */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- /* make sure additional input data as large as expected */
- data_size = le32_to_cpu(msg->data_size);
- if (mp->msg_len != (data_size + sizeof(*msg))) {
- HAIL_INFO(&srv_log, "PUT len mismatch: msg len %zu, "
- "wanted %zu + %u (== %zu)",
- mp->msg_len,
- sizeof(*msg), data_size, data_size + sizeof(*msg));
- resp_rc = CLE_BAD_PKT;
- goto err_out_noabort;
- }
-
- fh = le64_to_cpu(msg->fh);
-
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
dbenv->err(dbenv, rc, "DB_ENV->txn_begin");
@@ -774,7 +725,7 @@ void msg_put(struct msg_params *mp)
}
/* read handle from db */
- rc = cldb_handle_get(txn, sess->sid, fh, &h, 0);
+ rc = cldb_handle_get(txn, sess->sid, put->fh, &h, 0);
if (rc) {
resp_rc = CLE_FH_INVAL;
goto err_out;
@@ -797,14 +748,14 @@ void msg_put(struct msg_params *mp)
}
/* store contig. data area in db */
- mem = (msg + 1);
- rc = cldb_data_put(txn, inum, mem, data_size, 0);
+ rc = cldb_data_put(txn, inum,
+ put->data.data_val, put->data.data_len, 0);
if (rc) {
resp_rc = CLE_DB_ERR;
goto err_out;
}
- inode->size = cpu_to_le32(data_size);
+ inode->size = cpu_to_le32(put->data.data_len);
/* update inode */
rc = inode_touch(txn, inode);
@@ -820,7 +771,7 @@ void msg_put(struct msg_params *mp)
goto err_out_noabort;
}
- resp_ok(sess, mp->msg);
+ sess_sendresp_generic(sess, CLE_OK);
free(h);
free(inode);
@@ -831,31 +782,23 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_put txn abort");
err_out_noabort:
- resp_err(sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(h);
free(inode);
}
-void msg_close(struct msg_params *mp)
+void msg_close(struct session *sess, const void *v)
{
- const struct cld_msg_close *msg = mp->msg;
- uint64_t fh;
+ const struct cld_msg_close *close = v;
int rc;
enum cle_err_codes resp_rc = CLE_OK;
struct raw_handle *h = NULL;
cldino_t lock_inum = 0;
bool waiter = false;
- struct session *sess = mp->sess;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- fh = le64_to_cpu(msg->fh);
-
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
dbenv->err(dbenv, rc, "DB_ENV->txn_begin");
@@ -864,7 +807,7 @@ void msg_close(struct msg_params *mp)
}
/* read handle from db */
- rc = cldb_handle_get(txn, sess->sid, fh, &h, DB_RMW);
+ rc = cldb_handle_get(txn, sess->sid, close->fh, &h, DB_RMW);
if (rc) {
if (rc == DB_NOTFOUND)
resp_rc = CLE_FH_INVAL;
@@ -877,14 +820,15 @@ void msg_close(struct msg_params *mp)
lock_inum = cldino_from_le(h->inum);
/* delete handle from db */
- rc = cldb_handle_del(txn, sess->sid, fh);
+ rc = cldb_handle_del(txn, sess->sid, close->fh);
if (rc) {
resp_rc = CLE_DB_ERR;
goto err_out;
}
/* remove locks, if any */
- rc = session_remove_locks(txn, sess->sid, fh, lock_inum, &waiter);
+ rc = session_remove_locks(txn, sess->sid,
+ close->fh, lock_inum, &waiter);
if (rc) {
resp_rc = CLE_DB_ERR;
goto err_out;
@@ -906,7 +850,7 @@ void msg_close(struct msg_params *mp)
goto err_out_noabort;
}
- resp_ok(sess, mp->msg);
+ sess_sendresp_generic(sess, CLE_OK);
free(h);
return;
@@ -915,16 +859,15 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_close txn abort");
err_out_noabort:
- resp_err(sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(h);
}
-void msg_del(struct msg_params *mp)
+void msg_del(struct session *sess, const void *v)
{
- const struct cld_msg_del *msg = mp->msg;
+ const struct cld_msg_del *del = v;
enum cle_err_codes resp_rc = CLE_OK;
int rc, name_len;
- const char *name;
struct pathname_info pinfo;
struct raw_inode *parent = NULL, *ino = NULL;
void *parent_data = NULL;
@@ -937,23 +880,13 @@ void msg_del(struct msg_params *mp)
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- name_len = le16_to_cpu(msg->name_len);
-
- if (mp->msg_len < (sizeof(*msg) + name_len))
- return;
-
- name = mp->msg + sizeof(*msg);
-
- if (!valid_inode_name(name, name_len) || (name_len < 2)) {
+ name_len = strlen(del->inode_name);
+ if (!valid_inode_name(del->inode_name, name_len) || (name_len < 2)) {
resp_rc = CLE_NAME_INVAL;
goto err_out_noabort;
}
- pathname_parse(name, name_len, &pinfo);
+ pathname_parse(del->inode_name, name_len, &pinfo);
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
@@ -979,7 +912,8 @@ void msg_del(struct msg_params *mp)
}
/* read inode to be deleted */
- rc = cldb_inode_get_byname(txn, name, name_len, &ino, false, 0);
+ rc = cldb_inode_get_byname(txn, del->inode_name, name_len,
+ &ino, false, 0);
if (rc) {
if (rc == DB_NOTFOUND)
resp_rc = CLE_NAME_INVAL;
@@ -1088,7 +1022,7 @@ void msg_del(struct msg_params *mp)
goto err_out_noabort;
}
- resp_ok(mp->sess, mp->msg);
+ sess_sendresp_generic(sess, CLE_OK);
free(ino);
free(parent);
free(parent_data);
@@ -1099,31 +1033,23 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_del txn abort");
err_out_noabort:
- resp_err(mp->sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(ino);
free(parent);
free(parent_data);
}
-void msg_unlock(struct msg_params *mp)
+void msg_unlock(struct session *sess, const void *v)
{
- const struct cld_msg_unlock *msg = mp->msg;
- uint64_t fh;
+ const struct cld_msg_unlock *unlock = v;
struct raw_handle *h = NULL;
cldino_t inum;
int rc;
enum cle_err_codes resp_rc = CLE_OK;
uint32_t omode;
- struct session *sess = mp->sess;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- fh = le64_to_cpu(msg->fh);
-
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
dbenv->err(dbenv, rc, "DB_ENV->txn_begin");
@@ -1132,7 +1058,7 @@ void msg_unlock(struct msg_params *mp)
}
/* read handle from db */
- rc = cldb_handle_get(txn, sess->sid, fh, &h, 0);
+ rc = cldb_handle_get(txn, sess->sid, unlock->fh, &h, 0);
if (rc) {
resp_rc = CLE_FH_INVAL;
goto err_out;
@@ -1147,7 +1073,7 @@ void msg_unlock(struct msg_params *mp)
}
/* attempt to given lock on filehandle */
- rc = cldb_lock_del(txn, sess->sid, fh, inum);
+ rc = cldb_lock_del(txn, sess->sid, unlock->fh, inum);
if (rc) {
resp_rc = CLE_LOCK_INVAL;
goto err_out;
@@ -1160,7 +1086,7 @@ void msg_unlock(struct msg_params *mp)
goto err_out_noabort;
}
- resp_ok(sess, mp->msg);
+ sess_sendresp_generic(sess, CLE_OK);
free(h);
return;
@@ -1169,30 +1095,22 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_unlock txn abort");
err_out_noabort:
- resp_err(sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(h);
}
-void msg_lock(struct msg_params *mp, bool wait)
+void msg_lock(struct session *sess, const void *v)
{
- const struct cld_msg_lock *msg = mp->msg;
- uint64_t fh;
+ const struct cld_msg_lock *lock = v;
+ bool wait = (sess->msg_type == CMT_LOCK);
struct raw_handle *h = NULL;
cldino_t inum;
int rc;
enum cle_err_codes resp_rc = CLE_OK;
- uint32_t lock_flags, omode;
+ uint32_t omode;
bool acquired = false;
DB_ENV *dbenv = cld_srv.cldb.env;
DB_TXN *txn;
- struct session *sess = mp->sess;
-
- /* make sure input data as large as expected */
- if (mp->msg_len < sizeof(*msg))
- return;
-
- fh = le64_to_cpu(msg->fh);
- lock_flags = le32_to_cpu(msg->flags);
rc = dbenv->txn_begin(dbenv, NULL, &txn, 0);
if (rc) {
@@ -1202,7 +1120,7 @@ void msg_lock(struct msg_params *mp, bool wait)
}
/* read handle from db */
- rc = cldb_handle_get(txn, sess->sid, fh, &h, 0);
+ rc = cldb_handle_get(txn, sess->sid, lock->fh, &h, 0);
if (rc) {
resp_rc = CLE_FH_INVAL;
goto err_out;
@@ -1217,8 +1135,8 @@ void msg_lock(struct msg_params *mp, bool wait)
}
/* attempt to add lock */
- rc = cldb_lock_add(txn, sess->sid, fh, inum,
- lock_flags & CLF_SHARED, wait, &acquired);
+ rc = cldb_lock_add(txn, sess->sid, lock->fh, inum,
+ lock->flags & CLF_SHARED, wait, &acquired);
if (rc) {
if (rc == DB_KEYEXIST)
resp_rc = CLE_LOCK_CONFLICT;
@@ -1241,7 +1159,7 @@ void msg_lock(struct msg_params *mp, bool wait)
}
/* lock was acquired immediately */
- resp_ok(mp->sess, mp->msg);
+ sess_sendresp_generic(sess, CLE_OK);
free(h);
return;
@@ -1250,7 +1168,7 @@ err_out:
if (rc)
dbenv->err(dbenv, rc, "msg_lock txn abort");
err_out_noabort:
- resp_err(mp->sess, mp->msg, resp_rc);
+ sess_sendresp_generic(sess, resp_rc);
free(h);
}
diff --git a/server/server.c b/server/server.c
index e7cff1e..3b15891 100644
--- a/server/server.c
+++ b/server/server.c
@@ -37,6 +37,7 @@
#include <openssl/hmac.h>
#include <cld-private.h>
#include "cld.h"
+#include <cld_fmt.h>
#define PROGRAM_NAME "cld"
@@ -125,11 +126,9 @@ int udp_tx(int sock_fd, struct sockaddr *addr, socklen_t addr_len,
{
ssize_t src;
- HAIL_DEBUG(&srv_log, "udp_tx, fd %d", sock_fd);
-
src = sendto(sock_fd, data, data_len, 0, addr, addr_len);
if (src < 0 && errno != EAGAIN)
- applog(LOG_ERR, "udp_tx sendto (fd %d, data_len %u): %s",
+ HAIL_ERR(&srv_log, "udp_tx sendto (fd %d, data_len %u): %s",
sock_fd, (unsigned int) data_len,
strerror(errno));
@@ -139,357 +138,415 @@ int udp_tx(int sock_fd, struct sockaddr *addr, socklen_t addr_len,
return 0;
}
-void resp_copy(struct cld_msg_resp *resp, const struct cld_msg_hdr *src)
+static int udp_rx_handle(struct session *sess,
+ void (*msg_handler)(struct session *sess, const void *),
+ xdrproc_t xdrproc, void *xdrdata)
{
- memcpy(&resp->hdr, src, sizeof(*src));
- resp->code = 0;
- resp->rsv = 0;
- resp->xid_in = src->xid;
-}
-
-void resp_err(struct session *sess,
- const struct cld_msg_hdr *src, enum cle_err_codes errcode)
-{
- struct cld_msg_resp resp;
-
- resp_copy(&resp, src);
- __cld_rand64(&resp.hdr.xid);
- resp.code = cpu_to_le32(errcode);
-
- if (sess->sock_fd <= 0) {
- applog(LOG_ERR, "Nul sock in response");
- return;
+ XDR xin;
+ xdrmem_create(&xin, sess->msg_buf, sess->msg_buf_len, XDR_DECODE);
+ if (!xdrproc(&xin, xdrdata)) {
+ HAIL_DEBUG(&srv_log, "%s: couldn't parse message of type %s\n",
+ __func__, __cld_msg_type_to_str(sess->msg_type));
+ xdr_destroy(&xin);
+ return CLE_BAD_PKT;
}
-
- sess_sendmsg(sess, &resp, sizeof(resp), NULL, NULL);
+ msg_handler(sess, xdrdata);
+ xdr_free(xdrproc, xdrdata);
+ xdr_destroy(&xin);
+ return 0;
}
-void resp_ok(struct session *sess, const struct cld_msg_hdr *src)
+/** Recieve a UDP packet
+ *
+ * @param sock_fd The UDP socket we received the packet on
+ * @param cli Client address data
+ * @param info Packet information
+ * @param raw_pkt The raw packet buffer
+ * @param raw_len Length of the raw packet buffer
+ *
+ * @return An error code if we should send an error message
+ * response. CLE_OK if we are done.
+ */
+static enum cle_err_codes udp_rx(int sock_fd,
+ const struct client *cli, struct pkt_info *info,
+ const char *raw_pkt, size_t raw_len)
{
- resp_err(sess, src, CLE_OK);
-}
+ struct cld_pkt_hdr *pkt = info->pkt;
+ struct session *sess = info->sess;
-static const char *user_key(const char *user)
-{
- /* TODO: better auth scheme.
- * for now, use simple username==password auth scheme
- */
- if (!user || !*user ||
- (strnlen(user, 32) >= 32))
- return NULL;
+ if (sess) {
+ size_t msg_len;
- return user; /* our secret key */
-}
+ /* advance sequence id's and update last-contact timestamp */
+ sess->last_contact = current_time.tv_sec;
+ sess->sock_fd = sock_fd;
-static bool authcheck(const struct cld_packet *pkt, size_t pkt_len)
-{
- const char *key;
- unsigned char md[SHA_DIGEST_LENGTH];
- unsigned int md_len = 0;
- const void *p = pkt;
+ if (info->type != CMT_ACK) {
+ /* received message - update session */
+ sess->next_seqid_in++;
+ }
- key = user_key(pkt->user);
- if (!key)
- return false;
+ /* copy message fragment into reassembly buffer */
+ if (pkt->mi.order & CLD_PKT_IS_FIRST) {
+ sess->msg_type = info->type;
+ sess->msg_xid = info->xid;
+ sess->msg_buf_len = 0;
+ }
+ msg_len = raw_len - info->hdr_len - CLD_PKT_FTR_LEN;
+ if ((sess->msg_buf_len + msg_len) > CLD_MAX_MSG_SZ)
+ return CLE_BAD_PKT;
- HMAC(EVP_sha1(), key, strlen(key), p, pkt_len - SHA_DIGEST_LENGTH,
- md, &md_len);
+ memcpy(sess->msg_buf + sess->msg_buf_len,
+ raw_pkt + info->hdr_len, msg_len);
+ sess->msg_buf_len += msg_len;
+ }
- if (md_len != SHA_DIGEST_LENGTH)
- return false; /* BUG */
+ if (!(pkt->mi.order & CLD_PKT_IS_LAST)) {
+ struct cld_msg_ack_frag ack;
+ ack.seqid = info->seqid;
- if (memcmp(md, p + (pkt_len - SHA_DIGEST_LENGTH), SHA_DIGEST_LENGTH))
- return false;
+ /* transmit ack-partial-msg response (once, without retries) */
+ simple_sendmsg(sock_fd, cli, pkt->sid,
+ pkt->user, 0xdeadbeef,
+ (xdrproc_t)xdr_cld_msg_ack_frag, (void *)&ack,
+ CMT_ACK_FRAG);
+ return CLE_OK;
+ }
- return true;
+ /* Handle a complete message */
+ switch (info->type) {
+ case CMT_GET:
+ /* fall through */
+ case CMT_GET_META: {
+ struct cld_msg_get get = {0};
+ return udp_rx_handle(sess, msg_get,
+ (xdrproc_t)xdr_cld_msg_get, &get);
+ }
+ case CMT_OPEN: {
+ struct cld_msg_open open = {0};
+ return udp_rx_handle(sess, msg_open,
+ (xdrproc_t)xdr_cld_msg_open, &open);
+ }
+ case CMT_PUT: {
+ struct cld_msg_put put = {0};
+ return udp_rx_handle(sess, msg_put,
+ (xdrproc_t)xdr_cld_msg_put, &put);
+ }
+ case CMT_CLOSE: {
+ struct cld_msg_close close = {0};
+ return udp_rx_handle(sess, msg_close,
+ (xdrproc_t)xdr_cld_msg_close, &close);
+ }
+ case CMT_DEL: {
+ struct cld_msg_del del = {0};
+ return udp_rx_handle(sess, msg_del,
+ (xdrproc_t)xdr_cld_msg_del, &del);
+ }
+ case CMT_UNLOCK: {
+ struct cld_msg_unlock unlock = {0};
+ return udp_rx_handle(sess, msg_unlock,
+ (xdrproc_t)xdr_cld_msg_unlock, &unlock);
+ }
+ case CMT_TRYLOCK:
+ /* fall through */
+ case CMT_LOCK: {
+ struct cld_msg_lock lock = {0};
+ return udp_rx_handle(sess, msg_lock,
+ (xdrproc_t)xdr_cld_msg_lock, &lock);
+ }
+ case CMT_ACK:
+ msg_ack(sess, info->seqid);
+ return 0;
+ case CMT_NOP:
+ sess_sendresp_generic(sess, CLE_OK);
+ return 0;
+ case CMT_NEW_SESS:
+ msg_new_sess(sock_fd, cli, info);
+ return 0;
+ case CMT_END_SESS:
+ msg_end_sess(sess, info->xid);
+ return 0;
+ default:
+ HAIL_DEBUG(&srv_log, "%s: unexpected packet of type %s\n",
+ __func__, __cld_msg_type_to_str(info->type));
+ /* do nothing */
+ return 0;
+ }
}
-bool authsign(struct cld_packet *pkt, size_t pkt_len)
+/** Parse a packet's header. Verify that the magic number is correct.
+ *
+ * @param raw_pkt Pointer to the packet data
+ * @param raw_len Length of the raw data
+ * @param pkt (out param) the packet header
+ * @param hdr_len (out param) the length of the packet header
+ *
+ * @return true on success; false if this packet is garbage
+ */
+static bool parse_pkt_header(const char *raw_pkt, int raw_len,
+ struct cld_pkt_hdr *pkt, ssize_t *hdr_len)
{
- const char *key;
- unsigned char md[SHA_DIGEST_LENGTH];
- unsigned int md_len = 0;
- void *buf = pkt;
+ XDR xin;
+ static const char * const magic = CLD_PKT_MAGIC;
- key = user_key(pkt->user);
- if (!key)
+ if (raw_len <= CLD_PKT_FTR_LEN) {
+ HAIL_DEBUG(&srv_log, "%s: packet is too short: only "
+ "%d bytes\n", __func__, raw_len);
return false;
+ }
+ xdrmem_create(&xin, (void *)raw_pkt, raw_len - CLD_PKT_FTR_LEN,
+ XDR_DECODE);
+ memset(pkt, 0, sizeof(*pkt));
+ if (!xdr_cld_pkt_hdr(&xin, pkt)) {
+ HAIL_DEBUG(&srv_log, "%s: couldn't parse packet header\n",
+ __func__);
+ xdr_destroy(&xin);
+ return false;
+ }
+ *hdr_len = xdr_getpos(&xin);
+ xdr_destroy(&xin);
- HMAC(EVP_sha1(), key, strlen(key), buf, pkt_len - SHA_DIGEST_LENGTH,
- md, &md_len);
-
- if (md_len != SHA_DIGEST_LENGTH)
- applog(LOG_ERR, "authsign BUG: md_len != SHA_DIGEST_LENGTH");
-
- memcpy(buf + pkt_len - SHA_DIGEST_LENGTH, md, SHA_DIGEST_LENGTH);
+ if (memcmp((void *)&pkt->magic, magic, sizeof(pkt->magic))) {
+ HAIL_DEBUG(&srv_log, "%s: bad magic number\n", __func__);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)pkt);
+ return false;
+ }
return true;
}
-const char *opstr(enum cld_msg_ops op)
+/** Look up some information about a packet, including its session and the
+ * type of message it carries.
+ *
+ * @param pkt The packet's header
+ * @param raw_pkt Pointer to the raw packet data
+ * @param raw_len Length of the raw packet data
+ * @param info (out param) Information about the packet
+ *
+ * @return true on success; false if this packet is garbage
+ */
+static bool get_pkt_info(struct cld_pkt_hdr *pkt,
+ const char *raw_pkt, size_t raw_len,
+ size_t hdr_len, struct pkt_info *info)
{
- switch (op) {
- case cmo_nop: return "cmo_nop";
- case cmo_new_sess: return "cmo_new_sess";
- case cmo_open: return "cmo_open";
- case cmo_get_meta: return "cmo_get_meta";
- case cmo_get: return "cmo_get";
- case cmo_put: return "cmo_put";
- case cmo_close: return "cmo_close";
- case cmo_del: return "cmo_del";
- case cmo_lock: return "cmo_lock";
- case cmo_unlock: return "cmo_unlock";
- case cmo_trylock: return "cmo_trylock";
- case cmo_ack: return "cmo_ack";
- case cmo_end_sess: return "cmo_end_sess";
- case cmo_ping: return "cmo_ping";
- case cmo_not_master: return "cmo_not_master";
- case cmo_event: return "cmo_event";
- case cmo_ack_frag: return "cmo_ack_frag";
- default: return "(unknown)";
+ struct cld_pkt_ftr *foot;
+ struct session *s;
+
+ memset(info, 0, sizeof(info));
+ info->pkt = pkt;
+ info->sess = s = g_hash_table_lookup(cld_srv.sessions, &pkt->sid);
+ foot = (struct cld_pkt_ftr *)
+ (raw_pkt + (raw_len - CLD_PKT_FTR_LEN));
+ info->seqid = le64_to_cpu(foot->seqid);
+
+ if (pkt->mi.order & CLD_PKT_IS_FIRST) {
+ info->xid = pkt->mi.cld_pkt_msg_info_u.mi.xid;
+ info->type = pkt->mi.cld_pkt_msg_info_u.mi.type;
+ } else {
+ if (!s) {
+ HAIL_DEBUG(&srv_log, "%s: packet is not first, "
+ "but also not part of an existing session. "
+ "Protocol error.\n", __func__);
+ return false;
+ }
+ info->xid = s->msg_xid;
+ info->type = s->msg_type;
}
+ info->hdr_len = hdr_len;
+ return true;
}
-static void show_msg(const struct cld_msg_hdr *msg)
+/** Verify that the client session matches IP and username
+ *
+ * @param info Packet information
+ * @param cli Client address data
+ *
+ * @return 0 on success; error code otherwise
+ */
+static enum cle_err_codes validate_pkt_session(const struct pkt_info *info,
+ const struct client *cli)
{
- switch (msg->op) {
- case cmo_nop:
- case cmo_new_sess:
- case cmo_open:
- case cmo_get_meta:
- case cmo_get:
- case cmo_put:
- case cmo_close:
- case cmo_del:
- case cmo_lock:
- case cmo_unlock:
- case cmo_trylock:
- case cmo_ack:
- case cmo_end_sess:
- case cmo_ping:
- case cmo_not_master:
- case cmo_event:
- case cmo_ack_frag:
- HAIL_DEBUG(&srv_log, "msg: op %s, xid %llu",
- opstr(msg->op),
- (unsigned long long) le64_to_cpu(msg->xid));
- break;
+ struct session *sess = info->sess;
+
+ if (!sess) {
+ /* Packets that don't belong to a session must be new-session
+ * packets attempting to establish a session. */
+ if (info->type != CMT_NEW_SESS) {
+ HAIL_DEBUG(&srv_log, "%s: packet doesn't belong to a "
+ "session,but has type %d\n",
+ __func__, info->type);
+ return CLE_SESS_INVAL;
+ }
+ return 0;
}
-}
-
-static void udp_rx_msg(const struct client *cli, const struct cld_packet *pkt,
- const struct cld_msg_hdr *msg, struct msg_params *mp)
-{
- struct session *sess = mp->sess;
- if (srv_log.verbose)
- show_msg(msg);
+ if (info->type == CMT_NEW_SESS) {
+ HAIL_DEBUG(&srv_log, "%s: Tried to create a new session, "
+ "but a session with that ID already exists.\n",
+ __func__);
+ return CLE_SESS_EXISTS;
+ }
- switch(msg->op) {
- case cmo_nop:
- resp_ok(sess, msg);
- break;
+ /* verify that client session matches IP */
+ if ((sess->addr_len != cli->addr_len) ||
+ memcmp(&sess->addr, &cli->addr, sess->addr_len)) {
+ HAIL_DEBUG(&srv_log, "%s: sess->addr doesn't match packet "
+ "addr\n", __func__);
+ return CLE_SESS_INVAL;
+ }
- case cmo_new_sess: msg_new_sess(mp, cli); break;
- case cmo_end_sess: msg_end_sess(mp, cli); break;
- case cmo_open: msg_open(mp); break;
- case cmo_get: msg_get(mp, false); break;
- case cmo_get_meta: msg_get(mp, true); break;
- case cmo_put: msg_put(mp); break;
- case cmo_close: msg_close(mp); break;
- case cmo_del: msg_del(mp); break;
- case cmo_unlock: msg_unlock(mp); break;
- case cmo_lock: msg_lock(mp, true); break;
- case cmo_trylock: msg_lock(mp, false); break;
- case cmo_ack: msg_ack(mp); break;
+ /* verify that client session matches username */
+ if (strncmp(info->pkt->user, sess->user, CLD_MAX_USERNAME)) {
+ HAIL_DEBUG(&srv_log, "%s: session doesn't match packet's "
+ "username\n", __func__);
+ return CLE_SESS_INVAL;
+ }
- default:
- /* do nothing */
- break;
+ if (sess->dead) {
+ HAIL_DEBUG(&srv_log, "%s: packet session is dead\n",
+ __func__);
+ return CLE_SESS_INVAL;
}
+
+ return 0;
}
-static void pkt_ack_frag(int sock_fd,
- const struct client *cli,
- const struct cld_packet *pkt)
+/** Check a packet's cryptographic signature
+ *
+ * @param raw_pkt Pointer to the packet data
+ * @param raw_len Length of the raw data
+ * @param pkt the packet header
+ *
+ * @return 0 on success; error code otherwise
+ */
+static enum cle_err_codes validate_pkt_signature(const char *raw_pkt,
+ int raw_len, const struct cld_pkt_hdr *pkt)
{
- size_t alloc_len;
- struct cld_packet *outpkt;
- struct cld_msg_ack_frag *ack_msg;
-
- alloc_len = sizeof(*outpkt) + sizeof(*ack_msg) + SHA_DIGEST_LENGTH;
- outpkt = alloca(alloc_len);
- ack_msg = (struct cld_msg_ack_frag *) (outpkt + 1);
- memset(outpkt, 0, alloc_len);
-
- pkt_init_pkt(outpkt, pkt);
-
- memcpy(ack_msg->hdr.magic, CLD_MSG_MAGIC, CLD_MAGIC_SZ);
- __cld_rand64(&ack_msg->hdr.xid);
- ack_msg->hdr.op = cmo_ack_frag;
- ack_msg->seqid = pkt->seqid;
-
- authsign(outpkt, alloc_len);
-
- HAIL_DEBUG(&srv_log, "ack-partial-msg: "
- "sid " SIDFMT ", op %s, seqid %llu",
- SIDARG(outpkt->sid), opstr(ack_msg->hdr.op),
- (unsigned long long) le64_to_cpu(outpkt->seqid));
+ int ret;
+ struct cld_pkt_ftr *foot;
+
+ foot = (struct cld_pkt_ftr *)
+ (raw_pkt + (raw_len - CLD_PKT_FTR_LEN));
+ ret = __cld_authcheck(&srv_log, pkt->user,
+ raw_pkt, raw_len - SHA_DIGEST_LENGTH,
+ foot->sha);
+ if (ret) {
+ HAIL_DEBUG(&srv_log, "%s: bad HMAC signature on packet "
+ "ret = %d\n", __func__, ret);
+ return CLE_SIG_INVAL;
+ }
- /* transmit ack-partial-msg response (once, without retries) */
- udp_tx(sock_fd, (struct sockaddr *) &cli->addr, cli->addr_len,
- outpkt, alloc_len);
+ return 0;
}
-static void udp_rx(int sock_fd,
- const struct client *cli,
- const void *raw_pkt, size_t pkt_len)
+/** Check if this packet is a duplicate
+ *
+ * @param info Packet info
+ *
+ * @return true only if the packet is a duplicate
+ */
+static bool packet_is_dupe(const struct pkt_info *info)
{
- const struct cld_packet *pkt = raw_pkt;
- struct cld_packet *outpkt;
- const struct cld_msg_hdr *msg = (struct cld_msg_hdr *) (pkt + 1);
- struct session *sess = NULL;
- enum cle_err_codes resp_rc = CLE_OK;
- struct cld_msg_resp *resp;
- struct msg_params mp;
- size_t alloc_len;
- uint32_t pkt_flags;
- bool first_frag, last_frag, have_new_sess, have_ack, have_put;
-
- /* verify pkt data integrity and credentials via HMAC signature */
- if (!authcheck(pkt, pkt_len)) {
- resp_rc = CLE_SIG_INVAL;
- goto err_out;
- }
+ if (!info->sess)
+ return false;
+ if (info->type == CMT_ACK)
+ return false;
- pkt_flags = le32_to_cpu(pkt->flags);
- first_frag = pkt_flags & CPF_FIRST;
- last_frag = pkt_flags & CPF_LAST;
- have_new_sess = first_frag && (msg->op == cmo_new_sess);
- have_ack = first_frag && (msg->op == cmo_ack);
- have_put = first_frag && (msg->op == cmo_put);
-
- /* look up client session, verify it matches IP and username */
- sess = g_hash_table_lookup(cld_srv.sessions, pkt->sid);
- if (sess &&
- ((sess->addr_len != cli->addr_len) ||
- memcmp(&sess->addr, &cli->addr, sess->addr_len) ||
- strncmp(pkt->user, sess->user, CLD_MAX_USERNAME) ||
- sess->dead)) {
- resp_rc = CLE_SESS_INVAL;
- goto err_out;
+ /* Check sequence IDs to discover if this packet is a dupe */
+ if (info->seqid != info->sess->next_seqid_in) {
+ HAIL_DEBUG(&srv_log, "dropping dup with seqid %llu "
+ "(expected %llu).",
+ (unsigned long long) info->seqid,
+ (unsigned long long) info->sess->next_seqid_in);
+ return true;
}
- mp.sock_fd = sock_fd;
- mp.cli = cli;
- mp.sess = sess;
- mp.pkt = pkt;
- mp.msg = msg;
- mp.msg_len = pkt_len - sizeof(*pkt) - SHA_DIGEST_LENGTH;
-
- HAIL_DEBUG(&srv_log, "pkt: len %zu, seqid %llu, sid " SIDFMT ", "
- "flags %s%s, user %s",
- pkt_len, (unsigned long long) le64_to_cpu(pkt->seqid),
- SIDARG(pkt->sid),
- first_frag ? "F" : "", last_frag ? "L" : "",
- pkt->user);
-
- /* advance sequence id's and update last-contact timestamp */
- if (!have_new_sess) {
- if (!sess) {
- resp_rc = CLE_SESS_INVAL;
- goto err_out;
- }
-
- sess->last_contact = current_time.tv_sec;
- sess->sock_fd = sock_fd;
-
- if (!have_ack) {
- /* eliminate duplicates; do not return any response */
- if (le64_to_cpu(pkt->seqid) != sess->next_seqid_in) {
- HAIL_DEBUG(&srv_log, "dropping dup");
- return;
- }
-
- /* received message - update session */
- sess->next_seqid_in++;
- }
- } else {
- if (sess) {
- /* eliminate duplicates; do not return any response */
- if (le64_to_cpu(pkt->seqid) != sess->next_seqid_in) {
- HAIL_DEBUG(&srv_log, "dropping dup");
- return;
- }
+ return false;
+}
- resp_rc = CLE_SESS_EXISTS;
- goto err_out;
- }
+void simple_sendmsg(int fd, const struct client *cli,
+ uint64_t sid, const char *user, uint64_t seqid,
+ xdrproc_t xdrproc, const void *xdrdata, enum cld_msg_type type)
+{
+ XDR xhdr, xmsg;
+ struct cld_pkt_hdr pkt;
+ struct cld_pkt_msg_infos *infos;
+ struct cld_pkt_ftr *foot;
+ char *buf;
+ size_t msg_len, hdr_len, buf_len;
+
+ /* Set up the packet header */
+ memset(&pkt, 0, sizeof(cld_pkt_hdr));
+ memcpy(&pkt.magic, CLD_PKT_MAGIC, sizeof(pkt.magic));
+ pkt.sid = sid;
+ pkt.user = (char *)user;
+ pkt.mi.order = CLD_PKT_ORD_FIRST_LAST;
+ infos = &pkt.mi.cld_pkt_msg_info_u.mi;
+ __cld_rand64(&infos->xid);
+ infos->type = type;
+
+ /* Determine sizes */
+ msg_len = xdr_sizeof(xdrproc, (void *)xdrdata);
+ if (msg_len > CLD_MAX_MSG_SZ) {
+ HAIL_ERR(&srv_log, "%s: tried to put %d message bytes in a "
+ "single packet. Maximum message bytes per packet "
+ "is %d\n", __func__, msg_len, CLD_MAX_PKT_MSG_SZ);
+ return;
}
-
- /* copy message fragment into reassembly buffer */
- if (sess) {
- if (first_frag)
- sess->msg_buf_len = 0;
-
- if ((sess->msg_buf_len + mp.msg_len) > CLD_MAX_MSG_SZ) {
- resp_rc = CLE_BAD_PKT;
- goto err_out;
- }
-
- memcpy(&sess->msg_buf[sess->msg_buf_len], msg, mp.msg_len);
- sess->msg_buf_len += mp.msg_len;
-
- if (!last_frag) {
- pkt_ack_frag(sock_fd, cli, pkt);
- return;
- }
-
- mp.msg = msg = (struct cld_msg_hdr *) sess->msg_buf;
- mp.msg_len = sess->msg_buf_len;
-
- if ((srv_log.verbose > 1) && !first_frag)
- HAIL_DEBUG(&srv_log, " final message size %u",
- sess->msg_buf_len);
+ hdr_len = xdr_sizeof((xdrproc_t)xdr_cld_pkt_hdr, &pkt);
+ buf_len = msg_len + hdr_len + CLD_PKT_FTR_LEN;
+ buf = alloca(buf_len);
+
+ /* Serialize data */
+ xdrmem_create(&xhdr, buf, hdr_len, XDR_ENCODE);
+ if (!xdr_cld_pkt_hdr(&xhdr, &pkt)) {
+ xdr_destroy(&xhdr);
+ HAIL_ERR(&srv_log, "%s: xdr_cld_pkt_hdr failed\n",
+ __func__);
+ return;
}
+ xdr_destroy(&xhdr);
+ xdrmem_create(&xmsg, buf + hdr_len, msg_len, XDR_ENCODE);
+ if (!xdrproc(&xmsg, (void *)xdrdata)) {
+ xdr_destroy(&xmsg);
+ HAIL_ERR(&srv_log, "%s: xdrproc failed\n", __func__);
+ return;
+ }
+ xdr_destroy(&xmsg);
+
+ foot = (struct cld_pkt_ftr *)
+ (buf + (buf_len - SHA_DIGEST_LENGTH));
+ foot->seqid = cpu_to_le64(seqid);
+ __cld_authsign(&srv_log, pkt.user, buf, buf_len - SHA_DIGEST_LENGTH,
+ foot->sha);
+ udp_tx(fd, (struct sockaddr *) &cli->addr, cli->addr_len,
+ buf, buf_len);
+}
- if (last_frag)
- udp_rx_msg(cli, pkt, msg, &mp);
- return;
-
-err_out:
- /* transmit error response (once, without retries) */
- alloc_len = sizeof(*outpkt) + sizeof(*resp) + SHA_DIGEST_LENGTH;
- outpkt = alloca(alloc_len);
- resp = (struct cld_msg_resp *) (outpkt + 1);
- memset(outpkt, 0, alloc_len);
-
- pkt_init_pkt(outpkt, pkt);
-
- resp_copy(resp, msg);
- resp->code = cpu_to_le32(resp_rc);
-
- authsign(outpkt, alloc_len);
-
- HAIL_DEBUG(&srv_log, "udp_rx err: "
- "sid " SIDFMT ", op %s, seqid %llu, code %d",
- SIDARG(outpkt->sid), opstr(resp->hdr.op),
- (unsigned long long) le64_to_cpu(outpkt->seqid),
- resp_rc);
-
- udp_tx(sock_fd, (struct sockaddr *) &cli->addr, cli->addr_len,
- outpkt, alloc_len);
+static void simple_sendresp(int sock_fd, const struct client *cli,
+ const struct pkt_info *info, enum cle_err_codes code)
+{
+ const struct cld_pkt_hdr *pkt = info->pkt;
+ struct cld_msg_generic_resp resp;
+ resp.code = code;
+ resp.xid_in = info->xid;
+
+ simple_sendmsg(sock_fd, cli, pkt->sid, pkt->user, info->seqid,
+ (xdrproc_t)xdr_cld_msg_generic_resp, (void *)&resp,
+ info->type);
}
static bool udp_srv_event(int fd, short events, void *userdata)
{
struct client cli;
char host[64];
- ssize_t rrc;
+ ssize_t rrc, hdr_len;
struct msghdr hdr;
struct iovec iov[2];
- uint8_t raw_pkt[CLD_RAW_MSG_SZ], ctl_msg[CLD_RAW_MSG_SZ];
- struct cld_packet *pkt = (struct cld_packet *) raw_pkt;
+ char raw_pkt[CLD_RAW_MSG_SZ], ctl_msg[CLD_RAW_MSG_SZ];
+ struct cld_pkt_hdr pkt;
+ struct pkt_info info;
+ enum cle_err_codes err;
memset(&cli, 0, sizeof(cli));
@@ -519,40 +576,52 @@ static bool udp_srv_event(int fd, short events, void *userdata)
HAIL_DEBUG(&srv_log, "client %s message (%d bytes)", host, (int) rrc);
- /* if it is complete garbage, drop immediately */
- if ((rrc < (sizeof(*pkt) + SHA_DIGEST_LENGTH)) ||
- (memcmp(pkt->magic, CLD_PKT_MAGIC, sizeof(pkt->magic)))) {
+ if (!parse_pkt_header(raw_pkt, rrc, &pkt, &hdr_len)) {
cld_srv.stats.garbage++;
- return true; /* continue main loop; do NOT terminate server */
+ return true;
}
- if (cld_srv.cldb.is_master && cld_srv.cldb.up)
- udp_rx(fd, &cli, raw_pkt, rrc);
-
- else {
- struct cld_packet *outpkt;
- struct cld_msg_hdr *msg = (struct cld_msg_hdr *) (pkt + 1);
- struct cld_msg_resp *resp;
- size_t alloc_len;
-
- alloc_len = sizeof(*outpkt) + sizeof(*resp) + SHA_DIGEST_LENGTH;
- outpkt = alloca(alloc_len);
- memset(outpkt, 0, alloc_len);
+ if (!get_pkt_info(&pkt, raw_pkt, rrc, hdr_len, &info)) {
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ cld_srv.stats.garbage++;
+ return true;
+ }
- pkt_init_pkt(outpkt, pkt);
+ if (packet_is_dupe(&info)) {
+ /* silently drop dupes */
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
+ }
- /* transmit not-master error msg */
- resp = (struct cld_msg_resp *) (outpkt + 1);
- resp_copy(resp, msg);
- resp->hdr.op = cmo_not_master;
+ err = validate_pkt_session(&info, &cli);
+ if (err) {
+ simple_sendresp(fd, &cli, &info, err);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
+ }
- authsign(outpkt, alloc_len);
+ err = validate_pkt_signature(raw_pkt, rrc, &pkt);
+ if (err) {
+ simple_sendresp(fd, &cli, &info, err);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
+ }
- udp_tx(fd, (struct sockaddr *) &cli.addr, cli.addr_len,
- outpkt, alloc_len);
+ if (!(cld_srv.cldb.is_master && cld_srv.cldb.up)) {
+ simple_sendmsg(fd, &cli, pkt.sid, pkt.user, 0xdeadbeef,
+ (xdrproc_t)xdr_void, NULL, CMT_NOT_MASTER);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
}
- return true; /* continue main loop; do NOT terminate server */
+ err = udp_rx(fd, &cli, &info, raw_pkt, rrc);
+ if (err) {
+ simple_sendresp(fd, &cli, &info, err);
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
+ }
+ xdr_free((xdrproc_t)xdr_cld_pkt_hdr, (char *)&pkt);
+ return true;
}
static void add_chkpt_timer(void)
@@ -586,7 +655,7 @@ static int net_write_port(const char *port_file, const char *port_str)
portf = fopen(port_file, "w");
if (portf == NULL) {
rc = errno;
- applog(LOG_INFO, "Cannot create port file %s: %s",
+ HAIL_INFO(&srv_log, "Cannot create port file %s: %s",
port_file, strerror(rc));
return -rc;
}
@@ -607,7 +676,7 @@ static void net_close(void)
pfd = &g_array_index(cld_srv.polls, struct pollfd, i);
if (pfd->fd >= 0) {
if (close(pfd->fd) < 0)
- applog(LOG_WARNING, "net_close(%d): %s",
+ HAIL_WARN(&srv_log, "net_close(%d): %s",
pfd->fd, strerror(errno));
pfd->fd = -1;
}
@@ -674,7 +743,8 @@ static int net_open_any(void)
if (getsockname(fd6, (struct sockaddr *) &addr6,
&addr_len) != 0) {
rc = errno;
- applog(LOG_ERR, "getsockname failed: %s", strerror(rc));
+ HAIL_ERR(&srv_log, "getsockname failed: %s",
+ strerror(rc));
return -rc;
}
port = ntohs(addr6.sin6_port);
@@ -696,17 +766,18 @@ static int net_open_any(void)
if (getsockname(fd4, (struct sockaddr *) &addr4,
&addr_len) != 0) {
rc = errno;
- applog(LOG_ERR, "getsockname failed: %s", strerror(rc));
+ HAIL_ERR(&srv_log, "getsockname failed: %s",
+ strerror(rc));
return -rc;
}
port = ntohs(addr4.sin_port);
}
- applog(LOG_INFO, "Listening on port %u", port);
+ HAIL_INFO(&srv_log, "Listening on port %u", port);
if (cld_srv.port_file) {
char portstr[7];
- snprintf(portstr, sizeof(portstr), "%u", port);
+ snprintf(portstr, sizeof(portstr), "%u\n", port);
return net_write_port(cld_srv.port_file, portstr);
}
return 0;
@@ -725,7 +796,7 @@ static int net_open_known(const char *portstr)
rc = getaddrinfo(NULL, portstr, &hints, &res0);
if (rc) {
- applog(LOG_ERR, "getaddrinfo(*:%s) failed: %s",
+ HAIL_ERR(&srv_log, "getaddrinfo(*:%s) failed: %s",
portstr, gai_strerror(rc));
rc = -EINVAL;
goto err_addr;
@@ -765,7 +836,7 @@ static int net_open_known(const char *portstr)
listen_serv, sizeof(listen_serv),
NI_NUMERICHOST | NI_NUMERICSERV);
- applog(LOG_INFO, "Listening on %s port %s",
+ HAIL_INFO(&srv_log, "Listening on %s port %s",
listen_host, listen_serv);
}
@@ -791,7 +862,7 @@ static int net_open(void)
static void segv_signal(int signo)
{
- applog(LOG_ERR, "SIGSEGV");
+ HAIL_ERR(&srv_log, "SIGSEGV");
exit(1);
}
@@ -806,7 +877,7 @@ static void stats_signal(int signo)
}
#define X(stat) \
- applog(LOG_INFO, "STAT %s %lu", #stat, cld_srv.stats.stat)
+ HAIL_INFO(&srv_log, "STAT %s %lu", #stat, cld_srv.stats.stat)
static void stats_dump(void)
{
@@ -1037,7 +1108,7 @@ int main (int argc, char *argv[])
*/
rc = main_loop();
- applog(LOG_INFO, "shutting down");
+ HAIL_INFO(&srv_log, "shutting down");
if (strict_free)
timer_del(&cld_srv.chkpt_timer);
@@ -1087,7 +1158,7 @@ static void ensure_root()
} else if (rc == DB_NOTFOUND) {
inode = cldb_inode_mem("/", sizeof("/")-1, CIFL_DIR, CLD_INO_ROOT);
if (!inode) {
- applog(LOG_CRIT, "Cannot allocate new root inode");
+ HAIL_CRIT(&srv_log, "Cannot allocate new root inode");
goto err_;
}
@@ -1098,7 +1169,7 @@ static void ensure_root()
rc = cldb_inode_put(txn, inode, 0);
if (rc) {
free(inode);
- applog(LOG_CRIT, "Cannot allocate new root inode");
+ HAIL_CRIT(&srv_log, "Cannot allocate new root inode");
goto err_;
}
diff --git a/server/session.c b/server/session.c
index 44cd65a..768af81 100644
--- a/server/session.c
+++ b/server/session.c
@@ -18,6 +18,7 @@
*/
#define _GNU_SOURCE
+
#include "cld-config.h"
#include <sys/socket.h>
@@ -29,17 +30,15 @@
#include <openssl/sha.h>
#include <cld-private.h>
#include "cld.h"
-
-struct session_outpkt;
+#include <cld_fmt.h>
struct session_outpkt {
struct session *sess;
- struct cld_packet *pkt;
+ char *pkt_data;
size_t pkt_len;
uint64_t next_retry;
- uint64_t src_seqid;
unsigned int refs;
void (*done_cb)(struct session_outpkt *);
@@ -62,26 +61,6 @@ uint64_t next_seqid_le(uint64_t *seq)
return rc;
}
-void pkt_init_pkt(struct cld_packet *dest, const struct cld_packet *src)
-{
- memset(dest, 0, sizeof(*dest));
- memcpy(dest->magic, CLD_PKT_MAGIC, CLD_MAGIC_SZ);
- dest->seqid = cpu_to_le64(0xdeadbeef);
- memcpy(dest->sid, src->sid, CLD_SID_SZ);
- dest->flags = cpu_to_le32(CPF_FIRST | CPF_LAST);
- strncpy(dest->user, src->user, CLD_MAX_USERNAME - 1);
-}
-
-static void pkt_init_sess(struct cld_packet *dest, struct session *sess)
-{
- memset(dest, 0, sizeof(*dest));
- memcpy(dest->magic, CLD_PKT_MAGIC, CLD_MAGIC_SZ);
- dest->seqid = next_seqid_le(&sess->next_seqid_out);
- memcpy(dest->sid, sess->sid, CLD_SID_SZ);
- dest->flags = 0;
- strncpy(dest->user, sess->user, CLD_MAX_USERNAME - 1);
-}
-
guint sess_hash(gconstpointer v)
{
const struct session *sess = v;
@@ -399,20 +378,6 @@ static void session_ping_done(struct session_outpkt *outpkt)
outpkt->sess->ping_open = false;
}
-static void session_ping(struct session *sess)
-{
- struct cld_msg_hdr resp;
-
- memset(&resp, 0, sizeof(resp));
- memcpy(resp.magic, CLD_MSG_MAGIC, CLD_MAGIC_SZ);
- __cld_rand64(&resp.xid);
- resp.op = cmo_ping;
-
- sess->ping_open = true;
-
- sess_sendmsg(sess, &resp, sizeof(resp), session_ping_done, NULL);
-}
-
static void session_timeout(struct timer *timer)
{
struct session *sess = timer->userdata;
@@ -426,8 +391,11 @@ static void session_timeout(struct timer *timer)
if (!sess->dead && (sess_expire > now)) {
if (!sess->ping_open &&
(sess_expire > (sess->last_contact + (CLD_SESS_TIMEOUT / 2) &&
- (sess->sock_fd > 0))))
- session_ping(sess);
+ (sess->sock_fd > 0)))) {
+ sess_sendmsg(sess,
+ (xdrproc_t)xdr_void, NULL, CMT_PING,
+ session_ping_done, NULL);
+ }
timer_add(&sess->timer, now + ((sess_expire - now) / 2) + 1);
return; /* timer added; do not time out session */
@@ -520,8 +488,8 @@ static struct session_outpkt *op_alloc(size_t pkt_len)
if (!op)
return NULL;
- op->pkt = calloc(1, pkt_len);
- if (!op->pkt) {
+ op->pkt_data = calloc(1, pkt_len);
+ if (!op->pkt_data) {
free(op);
return NULL;
}
@@ -543,14 +511,13 @@ static void op_unref(struct session_outpkt *op)
return;
}
- free(op->pkt);
+ free(op->pkt_data);
free(op);
}
static int sess_retry_output(struct session *sess, time_t *next_retry_out)
{
- GList *tmp, *tmp1;
- struct session_outpkt *op;
+ GList *tmp;
int rc = 0;
time_t next_retry = 0;
@@ -558,28 +525,25 @@ static int sess_retry_output(struct session *sess, time_t *next_retry_out)
tmp = sess->out_q;
while (tmp) {
- struct cld_packet *outpkt;
- struct cld_msg_hdr *outmsg;
-
- tmp1 = tmp;
+ struct session_outpkt *op;
+ op = tmp->data;
tmp = tmp->next;
- op = tmp1->data;
- outpkt = op->pkt;
- outmsg = (struct cld_msg_hdr *) (outpkt + 1);
-
if (!next_retry || (op->next_retry < next_retry))
*next_retry_out = next_retry = op->next_retry;
if (current_time.tv_sec < op->next_retry)
continue;
- HAIL_DEBUG(&srv_log, "retry: sid " SIDFMT ", op %s, seqid %llu",
- SIDARG(outpkt->sid), opstr(outmsg->op),
- (unsigned long long) le64_to_cpu(outpkt->seqid));
+ if (srv_log.verbose) {
+ char scratch[PKT_HDR_TO_STR_SCRATCH_LEN];
+ HAIL_DEBUG(&srv_log, "%s: retrying %s\n", __func__,
+ __cld_pkt_hdr_to_str(scratch, op->pkt_data,
+ op->pkt_len));
+ }
rc = udp_tx(sess->sock_fd, (struct sockaddr *) &sess->addr,
- sess->addr_len, op->pkt, op->pkt_len);
+ sess->addr_len, op->pkt_data, op->pkt_len);
if (rc)
break;
@@ -611,141 +575,155 @@ static void session_outq(struct session *sess, GList *new_pkts)
sess->out_q = g_list_concat(sess->out_q, new_pkts);
}
-bool sess_sendmsg(struct session *sess, const void *msg_, size_t msglen,
- void (*done_cb)(struct session_outpkt *),
- void *done_data)
+bool sess_sendmsg(struct session *sess,
+ xdrproc_t xdrproc, const void *xdrdata, enum cld_msg_type type,
+ void (*done_cb)(struct session_outpkt *), void *done_data)
{
- struct cld_packet *outpkt;
- unsigned int n_pkts, i;
- size_t pkt_len, msg_left = msglen;
- struct session_outpkt *pkts[CLD_MAX_PKT_MSG], *op;
- GList *tmp_root = NULL;
- const void *p;
- bool first_frag = true;
-
- n_pkts = (msglen / CLD_MAX_PKT_MSG_SZ);
- n_pkts += (msglen % CLD_MAX_PKT_MSG_SZ) ? 1 : 0;
-
- if (srv_log.verbose) {
- const struct cld_msg_hdr *hdr = msg_;
- const struct cld_msg_resp *rsp;
-
- switch (hdr->op) {
- /* This is the command set that gets to cldc_rx_generic */
- case cmo_nop:
- case cmo_close:
- case cmo_del:
- case cmo_lock:
- case cmo_unlock:
- case cmo_trylock:
- case cmo_put:
- case cmo_new_sess:
- case cmo_end_sess:
- case cmo_open:
- case cmo_get_meta:
- case cmo_get:
- rsp = (struct cld_msg_resp *) msg_;
- HAIL_DEBUG(&srv_log, "sendmsg: "
- "sid " SIDFMT ", op %s, msglen %u, code %u, "
- "xid %llu, xid_in %llu",
- SIDARG(sess->sid),
- opstr(hdr->op),
- (unsigned int) msglen,
- le32_to_cpu(rsp->code),
- (unsigned long long) le64_to_cpu(hdr->xid),
- (unsigned long long) le64_to_cpu(rsp->xid_in));
- break;
- default:
- HAIL_DEBUG(&srv_log, "sendmsg: "
- "sid " SIDFMT ", op %s, msglen %u",
- SIDARG(sess->sid),
- opstr(hdr->op),
- (unsigned int) msglen);
- }
- }
+ XDR xmsg;
+ size_t msg_rem, msg_len, msg_chunk_len;
+ char *msg_bytes, *msg_cur;
+ GList *tmp_list, *new_pkts = NULL;
+ int first, last;
- if (n_pkts > CLD_MAX_PKT_MSG)
+ /* Use XDR to serialize the message */
+ msg_len = xdr_sizeof(xdrproc, (void *)xdrdata);
+ if (msg_len > CLD_MAX_MSG_SZ)
+ return false;
+ msg_bytes = alloca(msg_len);
+ xdrmem_create(&xmsg, msg_bytes, msg_len, XDR_ENCODE);
+ if (!xdrproc(&xmsg, (void *)xdrdata)) {
+ xdr_destroy(&xmsg);
+ HAIL_ERR(&srv_log, "%s: xdrproc failed\n", __func__);
return false;
-
- /* pass 1: perform allocations */
- for (i = 0; i < n_pkts; i++) {
- pkts[i] = op = op_alloc(sizeof(*outpkt) +
- CLD_MAX_PKT_MSG_SZ +
- SHA_DIGEST_LENGTH);
- if (!op)
- goto err_out;
-
- tmp_root = g_list_append(tmp_root, op);
}
+ xdr_destroy(&xmsg);
+
+ /* Break the message into packets */
+ first = 1;
+ msg_rem = msg_len;
+ msg_cur = msg_bytes;
+ do {
+ XDR xout;
+ struct cld_pkt_hdr pkt;
+ size_t hdr_len;
+ struct session_outpkt *op;
- /* pass 2: fill packets */
- p = msg_;
- for (i = 0; i < n_pkts; i++) {
- struct cld_msg_hdr *outmsg;
- void *outmsg_mem;
- size_t copy_len;
-
- op = pkts[i];
-
- op->sess = sess;
-
- outpkt = op->pkt;
- pkt_len = op->pkt_len;
-
- outmsg_mem = (outpkt + 1);
- outmsg = outmsg_mem;
-
- /* init packet header */
- pkt_init_sess(outpkt, sess);
-
- if (first_frag) {
- first_frag = false;
- outpkt->flags |= cpu_to_le32(CPF_FIRST);
+ if (msg_rem <= CLD_MAX_PKT_MSG_SZ) {
+ msg_chunk_len = msg_rem;
+ last = 1;
+ } else {
+ msg_chunk_len = CLD_MAX_PKT_MSG_SZ;
+ last = 0;
}
- copy_len = MIN(pkt_len - sizeof(*outpkt) - SHA_DIGEST_LENGTH,
- msg_left);
- memcpy(outmsg_mem, p, copy_len);
-
- p += copy_len;
- msg_left -= copy_len;
-
- op->pkt_len =
- pkt_len = sizeof(*outpkt) + copy_len + SHA_DIGEST_LENGTH;
-
- if (!msg_left) {
- op->done_cb = done_cb;
- op->done_data = done_data;
-
- outpkt->flags |= cpu_to_le32(CPF_LAST);
+ /* Set up packet header */
+ memset(&pkt, 0, sizeof(pkt));
+ memcpy(&pkt.magic, CLD_PKT_MAGIC, sizeof(pkt.magic));
+ memcpy(&pkt.sid, sess->sid, CLD_SID_SZ);
+ pkt.user = sess->user;
+ if (first) {
+ struct cld_pkt_msg_infos *infos =
+ &pkt.mi.cld_pkt_msg_info_u.mi;
+ if (last)
+ pkt.mi.order = CLD_PKT_ORD_FIRST_LAST;
+ else
+ pkt.mi.order = CLD_PKT_ORD_FIRST;
+ __cld_rand64(&infos->xid);
+ infos->type = type;
+ } else {
+ if (last)
+ pkt.mi.order = CLD_PKT_ORD_LAST;
+ else
+ pkt.mi.order = CLD_PKT_ORD_MID;
}
+ /* Allocate space and initialize session_outpkt structure */
+ hdr_len = xdr_sizeof((xdrproc_t)xdr_cld_pkt_hdr, (void *)&pkt);
+ op = op_alloc(hdr_len + msg_chunk_len + CLD_PKT_FTR_LEN);
+ if (!op) {
+ HAIL_DEBUG(&srv_log, "%s: op_alloc failed\n",
+ __func__);
+ goto err_out;
+ }
+ op->sess = sess;
op->next_retry = current_time.tv_sec + CLD_RETRY_START;
+ op->done_cb = done_cb;
+ op->done_data = done_data;
+ xdrmem_create(&xout, op->pkt_data, hdr_len, XDR_ENCODE);
+ if (!xdr_cld_pkt_hdr(&xout, &pkt)) {
+ xdr_destroy(&xout);
+ HAIL_ERR(&srv_log, "%s: xdr_cld_pkt_hdr failed\n",
+ __func__);
+ goto err_out;
+ }
+ xdr_destroy(&xout);
+
+ /* Fill in data */
+ memcpy(op->pkt_data + hdr_len, msg_cur, msg_chunk_len);
+ msg_cur += msg_chunk_len;
+ msg_rem -= msg_chunk_len;
+ first = 0;
+
+ new_pkts = g_list_prepend(new_pkts, op);
+ } while (!last);
+
+ /* add sequence IDs and SHAs */
+ new_pkts = g_list_reverse(new_pkts);
+ for (tmp_list = g_list_first(new_pkts);
+ tmp_list;
+ tmp_list = g_list_next(tmp_list)) {
+ struct session_outpkt *op =
+ (struct session_outpkt *) tmp_list->data;
+ struct cld_pkt_ftr *foot = (struct cld_pkt_ftr *)
+ (op->pkt_data + (op->pkt_len - CLD_PKT_FTR_LEN));
+ int ret;
+
+ foot->seqid = next_seqid_le(&sess->next_seqid_out);
+ ret = __cld_authsign(&srv_log, sess->user,
+ op->pkt_data, op->pkt_len - SHA_DIGEST_LENGTH,
+ foot->sha);
+ if (ret)
+ goto err_out;
+ }
- if (!authsign(outpkt, pkt_len))
- goto err_out; /* FIXME: we free all pkts -- wrong! */
-
+ /* send packets */
+ for (tmp_list = g_list_first(new_pkts);
+ tmp_list;
+ tmp_list = g_list_next(tmp_list)) {
+ struct session_outpkt *op =
+ (struct session_outpkt *) tmp_list->data;
udp_tx(sess->sock_fd, (struct sockaddr *) &sess->addr,
- sess->addr_len, outpkt, pkt_len);
+ sess->addr_len, op->pkt_data, op->pkt_len);
}
- session_outq(sess, tmp_root);
+ session_outq(sess, new_pkts);
return true;
err_out:
- for (i = 0; i < n_pkts; i++)
- op_unref(pkts[i]);
- g_list_free(tmp_root);
+ for (tmp_list = g_list_first(new_pkts); tmp_list;
+ tmp_list = g_list_next(tmp_list)) {
+ struct session_outpkt *op;
+ op = (struct session_outpkt *)tmp_list->data;
+ op_unref(op);
+ }
+ g_list_free(new_pkts);
return false;
}
-void msg_ack(struct msg_params *mp)
+void sess_sendresp_generic(struct session *sess, enum cle_err_codes code)
+{
+ struct cld_msg_generic_resp resp;
+ resp.code = code;
+ resp.xid_in = sess->msg_xid;
+
+ sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_generic_resp,
+ (void *)&resp, sess->msg_type, NULL, NULL);
+}
+
+void msg_ack(struct session *sess, uint64_t seqid)
{
- struct cld_packet *outpkt;
- struct cld_msg_hdr *outmsg;
GList *tmp, *tmp1;
- struct session *sess = mp->sess;
struct session_outpkt *op;
if (!sess->out_q)
@@ -754,19 +732,22 @@ void msg_ack(struct msg_params *mp)
/* look through output queue */
tmp = sess->out_q;
while (tmp) {
+ uint64_t op_seqid;
+ struct cld_pkt_ftr *foot;
tmp1 = tmp;
tmp = tmp->next;
op = tmp1->data;
- outpkt = op->pkt;
- outmsg = (struct cld_msg_hdr *) (outpkt + 1);
+ foot = (struct cld_pkt_ftr *)
+ (op->pkt_data + (op->pkt_len - CLD_PKT_FTR_LEN));
+ op_seqid = le64_to_cpu(foot->seqid);
/* if matching seqid found, we ack'd a message in out_q */
- if (mp->pkt->seqid != outpkt->seqid)
+ if (seqid != op_seqid)
continue;
HAIL_DEBUG(&srv_log, " expiring seqid %llu",
- (unsigned long long) le64_to_cpu(outpkt->seqid));
+ (unsigned long long) op_seqid);
/* remove and delete the ack'd msg; call ack'd callback */
sess->out_q = g_list_delete_link(sess->out_q, tmp1);
@@ -780,17 +761,17 @@ void msg_ack(struct msg_params *mp)
timer_del(&sess->retry_timer);
}
-void msg_new_sess(struct msg_params *mp, const struct client *cli)
+void msg_new_sess(int sock_fd, const struct client *cli,
+ const struct pkt_info *info)
{
+ const struct cld_pkt_hdr *pkt = info->pkt;
DB *db = cld_srv.cldb.sessions;
struct raw_session raw_sess;
struct session *sess = NULL;
DBT key, val;
int rc;
enum cle_err_codes resp_rc = CLE_OK;
- struct cld_msg_resp *resp;
- struct cld_packet *outpkt;
- size_t alloc_len;
+ struct cld_msg_generic_resp resp;
sess = session_new();
if (!sess) {
@@ -799,17 +780,17 @@ void msg_new_sess(struct msg_params *mp, const struct client *cli)
}
/* build raw_session database record */
- memcpy(sess->sid, mp->pkt->sid, sizeof(sess->sid));
+ memcpy(sess->sid, &pkt->sid, sizeof(sess->sid));
memcpy(&sess->addr, &cli->addr, sizeof(sess->addr));
- strncpy(sess->user, mp->pkt->user, sizeof(sess->user));
- sess->user[sizeof(sess->user) - 1] = 0;
+ snprintf(sess->user, sizeof(sess->user), "%s",
+ pkt->user);
- sess->sock_fd = mp->sock_fd;
+ sess->sock_fd = sock_fd;
sess->addr_len = cli->addr_len;
strncpy(sess->ipaddr, cli->addr_host, sizeof(sess->ipaddr));
sess->last_contact = current_time.tv_sec;
- sess->next_seqid_in = le64_to_cpu(mp->pkt->seqid) + 1;
+ sess->next_seqid_in = info->seqid + 1;
session_encode(&raw_sess, sess);
@@ -835,35 +816,33 @@ void msg_new_sess(struct msg_params *mp, const struct client *cli)
goto err_out;
}
+ HAIL_DEBUG(&srv_log, "%s: created new session " SIDFMT " with "
+ "sess->next_seqid_in = %llu\n",
+ __func__, SIDARG(sess->sid), sess->next_seqid_in);
g_hash_table_insert(cld_srv.sessions, sess->sid, sess);
/* begin session timer */
timer_add(&sess->timer, time(NULL) + (CLD_SESS_TIMEOUT / 2));
- resp_ok(sess, mp->msg);
+ /* send new-sess reply */
+ resp.code = CLE_OK;
+ resp.xid_in = info->xid;
+ sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_generic_resp,
+ (void *)&resp, CMT_NEW_SESS, NULL, NULL);
+
return;
err_out:
session_free(sess, true);
- alloc_len = sizeof(*outpkt) + sizeof(*resp) + SHA_DIGEST_LENGTH;
- outpkt = alloca(alloc_len);
- memset(outpkt, 0, alloc_len);
-
- pkt_init_pkt(outpkt, mp->pkt);
-
- resp = (struct cld_msg_resp *) (outpkt + 1);
- resp_copy(resp, mp->msg);
- resp->code = cpu_to_le32(resp_rc);
+ HAIL_DEBUG(&srv_log, "new_sess err: sid " SIDFMT ", op %s",
+ pkt->sid, __cld_msg_type_to_str(CMT_NEW_SESS));
- authsign(outpkt, alloc_len);
-
- HAIL_DEBUG(&srv_log, "new_sess err: sid " SIDFMT ", op %s, seqid %llu",
- SIDARG(outpkt->sid), opstr(resp->hdr.op),
- (unsigned long long) le64_to_cpu(outpkt->seqid));
-
- udp_tx(mp->sock_fd, (struct sockaddr *) &mp->cli->addr,
- mp->cli->addr_len, outpkt, alloc_len);
+ resp.code = resp_rc;
+ resp.xid_in = info->xid;
+ simple_sendmsg(sock_fd, cli, pkt->sid, pkt->user, 0xdeadbeef,
+ (xdrproc_t)xdr_cld_msg_generic_resp, (void *)&resp,
+ CMT_NEW_SESS);
HAIL_DEBUG(&srv_log, "NEW-SESS failed: %d", resp_rc);
}
@@ -873,18 +852,17 @@ static void end_sess_done(struct session_outpkt *outpkt)
session_trash(outpkt->sess);
}
-void msg_end_sess(struct msg_params *mp, const struct client *cli)
+void msg_end_sess(struct session *sess, uint64_t xid)
{
- struct session *sess = mp->sess;
- struct cld_msg_resp resp;
+ struct cld_msg_generic_resp resp;
/* do nothing; let message acknowledgement via
- * end_sess_done mark session dead
- */
-
- memset(&resp, 0, sizeof(resp));
- resp_copy(&resp, mp->msg);
- sess_sendmsg(sess, &resp, sizeof(resp), end_sess_done, NULL);
+ * end_sess_done mark session dead */
+ resp.code = CLE_OK;
+ resp.xid_in = xid;
+ sess_sendmsg(sess, (xdrproc_t)xdr_cld_msg_generic_resp,
+ &resp, CMT_END_SESS,
+ end_sess_done, NULL);
}
/*
diff --git a/test/Makefile.am b/test/Makefile.am
index 20dba03..d8048a7 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -1,6 +1,7 @@
-INCLUDES = -I$(top_srcdir)/include \
- @GLIB_CFLAGS@
+INCLUDES = -I$(top_srcdir)/include \
+ -I$(top_srcdir)/lib \
+ @GLIB_CFLAGS@
EXTRA_DIST = \
test.h \
diff --git a/test/load-file-event.c b/test/load-file-event.c
index 7b309b3..091e3a4 100644
--- a/test/load-file-event.c
+++ b/test/load-file-event.c
@@ -1,4 +1,3 @@
-
/*
* Copyright 2009 Red Hat, Inc.
*
@@ -20,7 +19,6 @@
/*
* Load a file from CLD.
*/
-
#define _GNU_SOURCE
#include "cld-config.h"
@@ -34,6 +32,8 @@
#include <cldc.h>
#include "test.h"
+#include "cld_fmt.h"
+
struct run {
struct cldc_udp *udp;
struct timer tmr_udp;
@@ -146,6 +146,8 @@ static int read_1_cb(struct cldc_call_opts *coptarg, enum cle_err_codes errc)
{
struct run *rp = coptarg->private;
struct cldc_call_opts copts;
+ char *data;
+ size_t data_len;
int rc;
if (errc != CLE_OK) {
@@ -153,12 +155,14 @@ static int read_1_cb(struct cldc_call_opts *coptarg, enum cle_err_codes errc)
exit(1);
}
- if (coptarg->u.get.size != TESTLEN) {
- fprintf(stderr, "Bad CLD file length %d\n", coptarg->u.get.size);
+ cldc_call_opts_get_data(coptarg, &data, &data_len);
+
+ if (data_len != TESTLEN) {
+ fprintf(stderr, "Bad CLD file length %d\n", data_len);
exit(1);
}
- if (memcmp(coptarg->u.get.buf, TESTSTR, TESTLEN)) {
+ if (memcmp(data, TESTSTR, TESTLEN)) {
fprintf(stderr, "Bad CLD file content\n");
exit(1);
}
diff --git a/test/save-file-event.c b/test/save-file-event.c
index 2b13bc8..03c6ca6 100644
--- a/test/save-file-event.c
+++ b/test/save-file-event.c
@@ -34,8 +34,6 @@
#include <libtimer.h>
#include "test.h"
-#define DATAMAX 10000
-
struct run {
struct cldc_udp *udp;
struct timer tmr_udp;
diff --git a/tools/cldcli.c b/tools/cldcli.c
index a06473d..e4d5c7c 100644
--- a/tools/cldcli.c
+++ b/tools/cldcli.c
@@ -267,6 +267,8 @@ static int cb_ls_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
struct cldc_call_opts copts = { NULL, };
struct cld_dirent_cur dc;
int rc, i;
+ char *data;
+ size_t data_len;
bool first = true;
if (errc != CLE_OK) {
@@ -274,8 +276,9 @@ static int cb_ls_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
write_from_thread(&cresp, sizeof(cresp));
return 0;
}
+ cldc_call_opts_get_data(copts_in, &data, &data_len);
- rc = cldc_dirent_count(copts_in->u.get.buf, copts_in->u.get.size);
+ rc = cldc_dirent_count(data, data_len);
if (rc < 0) {
write_from_thread(&cresp, sizeof(cresp));
return 0;
@@ -286,7 +289,7 @@ static int cb_ls_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
write_from_thread(&cresp, sizeof(cresp));
- cldc_dirent_cur_init(&dc, copts_in->u.get.buf, copts_in->u.get.size);
+ cldc_dirent_cur_init(&dc, data, data_len);
for (i = 0; i < rc; i++) {
struct ls_rec lsr;
@@ -343,6 +346,8 @@ static int cb_cat_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
{
struct cresp cresp = { .tcode = TC_FAILED, };
struct cldc_call_opts copts = { NULL, };
+ char *data;
+ size_t data_len;
if (errc != CLE_OK) {
errc_msg(&cresp, errc);
@@ -350,11 +355,13 @@ static int cb_cat_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
return 0;
}
+ cldc_call_opts_get_data(copts_in, &data, &data_len);
+
cresp.tcode = TC_OK;
- cresp.u.file_len = copts_in->u.get.size;
+ cresp.u.file_len = data_len;
write_from_thread(&cresp, sizeof(cresp));
- write_from_thread(copts_in->u.get.buf, copts_in->u.get.size);
+ write_from_thread(data, data_len);
/* FIXME: race; should wait until close succeeds/fails before
* returning any data. 'fh' may still be in use, otherwise.
@@ -387,6 +394,8 @@ static int cb_cp_cf_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
{
struct cresp cresp = { .tcode = TC_FAILED, };
struct cldc_call_opts copts = { NULL, };
+ char *data;
+ size_t data_len;
if (errc != CLE_OK) {
errc_msg(&cresp, errc);
@@ -394,11 +403,12 @@ static int cb_cp_cf_2(struct cldc_call_opts *copts_in, enum cle_err_codes errc)
return 0;
}
+ cldc_call_opts_get_data(copts_in, &data, &data_len);
cresp.tcode = TC_OK;
- cresp.u.file_len = copts_in->u.get.size;
+ cresp.u.file_len = data_len;
write_from_thread(&cresp, sizeof(cresp));
- write_from_thread(copts_in->u.get.buf, copts_in->u.get.size);
+ write_from_thread(data, data_len);
/* FIXME: race; should wait until close succeeds/fails before
* returning any data. 'fh' may still be in use, otherwise.
--
1.6.2.5
--
To unsubscribe from this list: send the line "unsubscribe hail-devel" in
the body of a message to majordomo@xxxxxxxxxxxxxxx
More majordomo info at http://vger.kernel.org/majordomo-info.html
