2 * Amanda, The Advanced Maryland Automatic Network Disk Archiver
3 * Copyright (c) 2008-2012 Zmanda, Inc. All Rights Reserved.
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 as published
7 * by the Free Software Foundation.
9 * This program is distributed in the hope that it will be useful, but
10 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
11 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 * Contact information: Zmanda Inc., 465 S. Mathilda Ave., Suite 300
19 * Sunnyvale, CA 94085, USA, or: http://www.zmanda.com
23 #include "ipc-binary.h"
25 struct ipc_binary_proto_t {
28 ipc_binary_cmd_t *cmds;
31 /* extra flag to indicate that an argument exists */
32 #define IPC_BINARY_EXISTS (1 << 7)
34 struct ipc_binary_cmd_t {
40 #define MSG_HDR_LEN 10
49 ipc_binary_buf_t *buf,
52 gsize new_len = buf->length + size;
54 /* allocate space in the buffer if necessary */
55 if (buf->offset + new_len > buf->size) {
56 if (buf->offset != 0 && new_len <= buf->size) {
58 buf->buf + buf->offset,
62 buf->size = buf->offset + new_len;
63 buf->buf = g_realloc(buf->buf, buf->size);
70 ipc_binary_buf_t *buf,
74 expand_buffer(buf, size);
76 g_memmove(buf->buf + buf->offset + buf->length, data, size);
82 ipc_binary_buf_t *buf,
85 g_assert(size <= buf->length);
96 ipc_binary_message_t *msg)
100 for (i = 0; i < msg->cmd->n_args; i++) {
101 if (msg->args[i].data == NULL
102 && (msg->cmd->arg_flags[i] & IPC_BINARY_EXISTS)
103 && !(msg->cmd->arg_flags[i] & IPC_BINARY_OPTIONAL)) {
104 g_debug("ipc-binary message missing mandatory arg %d", i);
113 * Creating a protocol
117 ipc_binary_proto_new(
120 ipc_binary_proto_t *prot = g_new(ipc_binary_proto_t, 1);
130 ipc_binary_proto_add_cmd(
131 ipc_binary_proto_t *proto,
134 g_assert(proto != NULL);
137 if (id >= proto->n_cmds) {
138 guint16 new_len = id+1;
141 proto->cmds = g_renew(ipc_binary_cmd_t, proto->cmds, new_len);
142 for (i = proto->n_cmds; i < new_len; i++) {
143 proto->cmds[i].n_args = 0;
144 proto->cmds[i].exists = FALSE;
145 proto->cmds[i].arg_flags = NULL;
147 proto->n_cmds = new_len;
150 /* make sure this command hasn't been defined already */
151 g_assert(!proto->cmds[id].exists);
152 proto->cmds[id].exists = TRUE;
154 return &proto->cmds[id];
158 ipc_binary_cmd_add_arg(
159 ipc_binary_cmd_t *cmd,
163 g_assert(cmd != NULL);
165 flags |= IPC_BINARY_EXISTS;
167 if (id >= cmd->n_args) {
168 guint16 new_len = id+1;
171 cmd->arg_flags = g_realloc(cmd->arg_flags, new_len);
172 for (i = cmd->n_args; i < new_len; i++) {
173 cmd->arg_flags[i] = 0;
175 cmd->n_args = new_len;
178 /* make sure this arg hasn't been defined already */
179 g_assert(cmd->arg_flags[id] == 0);
181 cmd->arg_flags[id] = flags;
188 ipc_binary_channel_t *
189 ipc_binary_new_channel(
190 ipc_binary_proto_t *proto)
192 ipc_binary_channel_t *chan;
194 chan = g_new0(ipc_binary_channel_t, 1);
201 ipc_binary_free_channel(
202 ipc_binary_channel_t *chan)
205 g_free(chan->in.buf);
208 g_free(chan->out.buf);
213 ipc_binary_message_t *
214 ipc_binary_new_message(
215 ipc_binary_channel_t *chan,
218 ipc_binary_message_t *msg = g_new0(ipc_binary_message_t, 1);
219 ipc_binary_cmd_t *cmd;
221 /* make sure this is a valid command */
222 g_assert(chan != NULL);
223 g_assert(cmd_id > 0 && cmd_id < chan->proto->n_cmds);
224 g_assert(chan->proto->cmds[cmd_id].exists);
225 cmd = &chan->proto->cmds[cmd_id];
229 msg->cmd_id = cmd_id;
230 msg->n_args = cmd->n_args;
231 msg->args = g_malloc0(sizeof(*(msg->args)) * cmd->n_args);
238 ipc_binary_message_t *msg,
242 gboolean take_memory)
244 /* make sure this arg has not already been set for this message */
245 g_assert(msg != NULL);
246 g_assert(data != NULL);
247 g_assert(arg_id > 0 && arg_id < msg->cmd->n_args);
248 g_assert(msg->cmd->arg_flags[arg_id] & IPC_BINARY_EXISTS);
249 g_assert(msg->args[arg_id].data == NULL);
251 if (size == 0 && msg->cmd->arg_flags[arg_id] & IPC_BINARY_STRING) {
252 size = strlen((gchar *)data);
256 data = g_memdup(data, size);
259 msg->args[arg_id].len = size;
260 msg->args[arg_id].data = data;
264 ipc_binary_free_message(
265 ipc_binary_message_t *msg)
269 g_assert(msg != NULL);
271 for (i = 0; i < msg->cmd->n_args; i++) {
272 gpointer data = msg->args[i].data;
281 ipc_binary_message_t *
282 ipc_binary_read_message(
283 ipc_binary_channel_t *chan,
286 ipc_binary_message_t *msg;
288 /* read data until we have a whole packet or until EOF */
289 while (!(msg = ipc_binary_poll_message(chan))) {
295 /* read directly into the buffer, instead of using ipc_binary_feed_data */
296 expand_buffer(&chan->in, 32768);
297 bytes = read(fd, chan->in.buf + chan->in.offset + chan->in.length, 32768);
302 /* got EOF; if there are bytes left over, this is EIO */
303 if (chan->in.length) {
304 g_warning("got EOF reading ipc-binary channel with %zd bytes un-processed",
311 /* add the data to the buffer */
312 chan->in.length += bytes;
320 ipc_binary_write_message(
321 ipc_binary_channel_t *chan,
323 ipc_binary_message_t *msg)
327 /* add the message to the queue */
328 ipc_binary_queue_message(chan, msg);
330 /* and write the outgoing buffer */
331 written = full_write(fd, chan->out.buf + chan->out.offset, chan->out.length);
332 consume_from_buffer(&chan->out, written);
334 if (written < chan->out.length) {
342 ipc_binary_feed_data(
343 ipc_binary_channel_t *chan,
347 add_to_buffer(&chan->in, size, data);
351 ipc_binary_data_transmitted(
352 ipc_binary_channel_t *chan,
355 consume_from_buffer(&chan->out, size);
359 get_guint16(guint8 **p) {
362 v = *((*p)++) | v << 8;
367 get_guint32(guint8 **p) {
370 v = *((*p)++) | v << 8;
371 v = *((*p)++) | v << 8;
372 v = *((*p)++) | v << 8;
377 ipc_binary_message_t *
378 ipc_binary_poll_message(
379 ipc_binary_channel_t *chan)
382 ipc_binary_message_t *msg;
388 if (chan->in.length < MSG_HDR_LEN) {
393 /* read out the pocket header, using shifts to avoid endian and alignment
394 * problems, and checking each one as we proceed */
395 p = (guint8 *)(chan->in.buf + chan->in.offset);
397 magic = get_guint16(&p);
399 if (magic != chan->proto->magic) {
400 g_debug("ipc-binary got invalid magic 0x%04x", (int)magic);
405 cmd_id = get_guint16(&p);
407 /* make sure this is a valid command */
408 if (cmd_id <= 0 || cmd_id >= chan->proto->n_cmds
409 || !chan->proto->cmds[cmd_id].exists) {
414 length = get_guint32(&p);
416 /* see if there's enough data in this buffer for a whole message */
417 if (length > chan->in.length) {
419 return NULL; /* whole packet isn't here yet */
422 n_args = get_guint16(&p);
424 /* looks legit -- start building a message */
425 msg = ipc_binary_new_message(chan, cmd_id);
427 /* get each of the arguments */
432 arglen = get_guint32(&p);
433 arg_id = get_guint16(&p);
435 if (arg_id <= 0 || arg_id >= msg->cmd->n_args
436 || !(msg->cmd->arg_flags[arg_id] & IPC_BINARY_EXISTS)
437 || msg->args[arg_id].data != NULL) {
438 g_debug("ipc-binary invalid or duplicate arg");
440 ipc_binary_free_message(msg);
444 /* properly terminate string args, but do not include the nul byte in
446 if (msg->cmd->arg_flags[arg_id] & IPC_BINARY_STRING) {
449 /* copy and terminate the string */
450 data = g_malloc(arglen+1);
451 g_memmove(data, p, arglen);
453 msg->args[arg_id].data = (gpointer)data;
454 msg->args[arg_id].len = arglen;
456 msg->args[arg_id].data = g_memdup(p, arglen);
457 msg->args[arg_id].len = arglen;
463 /* check that all mandatory args are here */
464 if (!all_args_present(msg)) {
466 ipc_binary_free_message(msg);
470 consume_from_buffer(&chan->in, length);
476 put_guint16(guint8 *p, guint16 v)
484 put_guint32(guint8 *p, guint32 v)
494 ipc_binary_queue_message(
495 ipc_binary_channel_t *chan G_GNUC_UNUSED,
496 ipc_binary_message_t *msg G_GNUC_UNUSED)
503 g_assert(all_args_present(msg));
505 /* calculate the length and make enough room in the buffer */
506 msg_len = MSG_HDR_LEN;
507 for (i = 0; i < msg->cmd->n_args; i++) {
508 if (msg->args[i].data) {
510 msg_len += msg->args[i].len + ARG_HDR_LEN;
513 expand_buffer(&chan->out, msg_len);
514 p = (guint8 *)(chan->out.buf + chan->out.offset);
516 /* write the packet */
517 p = put_guint16(p, chan->proto->magic);
518 p = put_guint16(p, msg->cmd_id);
519 p = put_guint32(p, msg_len);
520 p = put_guint16(p, n_args);
522 for (i = 0; i < msg->cmd->n_args; i++) {
523 if (!msg->args[i].data)
526 p = put_guint32(p, msg->args[i].len);
527 p = put_guint16(p, i);
529 g_memmove(p, msg->args[i].data, msg->args[i].len);
530 p += msg->args[i].len;
532 chan->out.length += msg_len;
534 ipc_binary_free_message(msg);