/*
- * Copyright (c) 2009, 2010 Zmanda, Inc. All Rights Reserved.
+ * Copyright (c) 2009-2012 Zmanda, Inc. All Rights Reserved.
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 as published
*/
#include "amanda.h"
+#include "event.h"
+#include "sockaddr-util.h"
#include "ndmpconnobj.h"
/*
*addrs = g_new0(DirectTCPAddr, naddrs+1);
for (i = 0; i < naddrs; i++) {
ndmp4_tcp_addr *na = &reply->connect_addr.ndmp4_addr_u.tcp_addr.tcp_addr_val[i];
- (*addrs)[i].ipv4 = na->ip_addr;
- (*addrs)[i].port = na->port;
+ (*addrs)[i].sin.sin_family = AF_INET;
+ (*addrs)[i].sin.sin_addr.s_addr = htonl(na->ip_addr);
+ SU_SET_PORT(addrs[i], na->port);
}
}
NDMP_FREE();
return TRUE;
}
+ndmp_connection_mover_connect(
+ NDMPConnection *self,
+ ndmp9_mover_mode mode,
+ DirectTCPAddr *addrs)
+{
+ unsigned int naddrs, i;
+ ndmp4_tcp_addr *na;
+
+ g_assert(!self->startup_err);
+
+ /* count addrs */
+ g_assert(addrs);
+ for (naddrs = 0; SU_GET_FAMILY(&addrs[naddrs]) != 0; naddrs++) ;
+
+ /* convert addrs to an ndmp4_tcp_addr */
+ na = g_new0(ndmp4_tcp_addr, naddrs);
+ for (i = 0; i < naddrs; i++) {
+ na[i].ip_addr = ntohl(addrs[i].sin.sin_addr.s_addr);
+ na[i].port = SU_GET_PORT(&addrs[i]);
+ }
+
+
+ NDMP_TRANS(self, ndmp4_mover_connect)
+ request->mode = mode;
+ request->addr.addr_type = NDMP4_ADDR_TCP;
+ request->addr.ndmp4_addr_u.tcp_addr.tcp_addr_len = naddrs;
+ request->addr.ndmp4_addr_u.tcp_addr.tcp_addr_val = na;
+ NDMP_CALL(self);
+ NDMP_FREE();
+ NDMP_END
+ return TRUE;
+}
+
gboolean
ndmp_connection_mover_abort(
NDMPConnection *self)
while (1) {
gboolean found = FALSE;
+ int fd;
+ SELECT_ARG_TYPE readset;
+ int nfound;
/* if any desired notifications have been received, then we're
* done */
return TRUE;
/* otherwise, wait for an incoming packet and handle it, then try
- * again */
+ * again. There's some select trickery here to avoid hogging the
+ * ndmlib_mutex - basically, we want to block as long as possible
+ * outside of the ndmlib_mutex critical section. This will also be
+ * useful to allow the wait to be aborted. */
+ fd = self->conn->chan.fd;
+ FD_ZERO(&readset);
+ FD_SET(fd, &readset);
+ nfound = select(fd+1, &readset, NULL, NULL, NULL);
+
+ /* fall on through, blind to any errors - presumably the same error
+ * condition will be caught by ndmconn_recv_nmb. */
+
g_static_mutex_lock(&ndmlib_mutex);
NDMOS_MACRO_ZEROFILL(&nmb);
nmb.protocol_version = NDMP4VER;
}
}
+typedef struct notify_data {
+ NDMPConnection *self;
+ ndmp9_data_halt_reason *data_halt_reason;
+ ndmp9_mover_halt_reason *mover_halt_reason;
+ ndmp9_mover_pause_reason *mover_pause_reason;
+ guint64 *mover_pause_seek_position;
+ GMutex *abort_mutex;
+ GCond *abort_cond;
+ int status;
+ event_handle_t *read_event;
+} notify_data;
+
+static void handle_notify(void *cookie);
+
+int
+ndmp_connection_wait_for_notify_with_cond(
+ NDMPConnection *self,
+ ndmp9_data_halt_reason *data_halt_reason,
+ ndmp9_mover_halt_reason *mover_halt_reason,
+ ndmp9_mover_pause_reason *mover_pause_reason,
+ guint64 *mover_pause_seek_position,
+ GMutex *abort_mutex,
+ GCond *abort_cond)
+{
+ struct ndmp_msg_buf nmb;
+ notify_data ndata;
+ gboolean found = FALSE;
+
+ ndata.self = self;
+ ndata.data_halt_reason= data_halt_reason;
+ ndata.mover_halt_reason= mover_halt_reason;
+ ndata.mover_pause_reason= mover_pause_reason;
+ ndata.mover_pause_seek_position = mover_pause_seek_position;
+ ndata.abort_mutex = abort_mutex;
+ ndata.abort_cond = abort_cond;
+ ndata.status = 2;
+
+ g_assert(!self->startup_err);
+
+ /* initialize output parameters */
+ if (data_halt_reason)
+ *data_halt_reason = NDMP4_DATA_HALT_NA;
+ if (mover_halt_reason)
+ *mover_halt_reason = NDMP4_MOVER_HALT_NA;
+ if (mover_pause_reason)
+ *mover_pause_reason = NDMP4_MOVER_PAUSE_NA;
+ if (mover_pause_seek_position)
+ *mover_pause_seek_position = 0;
+
+ /* if any desired notifications have been received, then we're
+ * done */
+ if (data_halt_reason && self->data_halt_reason) {
+ found = TRUE;
+ *data_halt_reason = self->data_halt_reason;
+ self->data_halt_reason = NDMP4_DATA_HALT_NA;
+ }
+
+ if (mover_halt_reason && self->mover_halt_reason) {
+ found = TRUE;
+ *mover_halt_reason = self->mover_halt_reason;
+ self->mover_halt_reason = NDMP4_MOVER_HALT_NA;
+ }
+
+ if (mover_pause_reason && self->mover_pause_reason) {
+ found = TRUE;
+ *mover_pause_reason = self->mover_pause_reason;
+ if (mover_pause_seek_position)
+ *mover_pause_seek_position = self->mover_pause_seek_position;
+ self->mover_pause_reason = NDMP4_MOVER_PAUSE_NA;
+ self->mover_pause_seek_position = 0;
+ }
+
+ if (found)
+ return TRUE;
+
+ /* otherwise, wait for an incoming packet and handle it, then try
+ * again. There's some select trickery here to avoid hogging the
+ * ndmlib_mutex - basically, we want to block as long as possible
+ * outside of the ndmlib_mutex critical section. This will also be
+ * useful to allow the wait to be aborted. */
+
+ ndata.read_event = event_register(self->conn->chan.fd,
+ EV_READFD, handle_notify, &ndata);
+
+ g_cond_wait(abort_cond, abort_mutex);
+
+ if (ndata.read_event) {
+ event_release(ndata.read_event);
+ }
+ if (ndata.status == 2) {
+ ndmp_connection_mover_abort(self);
+ ndmp_connection_mover_stop(self);
+ }
+ return ndata.status;
+
+}
+
+static void
+handle_notify(void *cookie)
+{
+ notify_data *ndata = cookie;
+ struct ndmp_msg_buf nmb;
+ gboolean found = FALSE;
+
+ g_mutex_lock(ndata->abort_mutex);
+
+ event_release(ndata->read_event);
+ ndata->read_event = NULL;
+
+ g_static_mutex_lock(&ndmlib_mutex);
+ NDMOS_MACRO_ZEROFILL(&nmb);
+ nmb.protocol_version = NDMP4VER;
+ ndata->self->last_rc = ndmconn_recv_nmb(ndata->self->conn, &nmb);
+ g_static_mutex_unlock(&ndmlib_mutex);
+
+ if (ndata->self->last_rc) {
+ /* (nothing to free) */
+ ndata->status = 1;
+ goto notify_done;
+ }
+
+ ndmconn_handle_notify(ndata->self, &nmb);
+
+
+ /* if any desired notifications have been received, then we're
+ * done */
+ if (ndata->data_halt_reason && ndata->self->data_halt_reason) {
+ found = TRUE;
+ *ndata->data_halt_reason = ndata->self->data_halt_reason;
+ ndata->self->data_halt_reason = NDMP4_DATA_HALT_NA;
+ }
+
+ if (ndata->mover_halt_reason && ndata->self->mover_halt_reason) {
+ found = TRUE;
+ *ndata->mover_halt_reason = ndata->self->mover_halt_reason;
+ ndata->self->mover_halt_reason = NDMP4_MOVER_HALT_NA;
+ }
+
+ if (ndata->mover_pause_reason && ndata->self->mover_pause_reason) {
+ found = TRUE;
+ *ndata->mover_pause_reason = ndata->self->mover_pause_reason;
+ if (ndata->mover_pause_seek_position)
+ *ndata->mover_pause_seek_position = ndata->self->mover_pause_seek_position;
+ ndata->self->mover_pause_reason = NDMP4_MOVER_PAUSE_NA;
+ ndata->self->mover_pause_seek_position = 0;
+ }
+
+ if (!found) {
+ ndata->read_event = event_register(ndata->self->conn->chan.fd,
+ EV_READFD, handle_notify, ndata);
+ g_mutex_unlock(ndata->abort_mutex);
+ return;
+ }
+
+ ndata->status = 0;
+notify_done:
+ g_cond_broadcast(ndata->abort_cond);
+ g_mutex_unlock(ndata->abort_mutex);
+}
/*
* Class Mechanics
*/
projects / debian / amanda / blobdiff