data.c

网络数据管理协议的开发

	memset((void*)&reply, 0, sizeof(reply));
	
	if (session->data.state != NDMP_DATA_STATE_IDLE)
	{
		Error(LOG_ERR,
			  "ndmpdDataConnect: invalid state to process connect request.\n");

		reply.error = NDMP_ILLEGAL_STATE_ERR;
		if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
			Error(LOG_ERR,
				  "ndmpdDataConnect: error sending ndmp_data_connect_reply.\n");
		return;
	}

	reply.error = NDMP_NO_ERR;

	switch (request->addr.addr_type)
	{
		case NDMP_ADDR_TCP:
		{
			struct sockaddr_in	sin;
			int					sock;
			int					flags;
		
			memset((void*)&sin, 0, sizeof(sin));
			sin.sin_family      = AF_INET;
			sin.sin_addr.s_addr = htonl(request->addr.ndmp_addr_u.tcp_addr.ip_addr);
			sin.sin_port        = htons(request->addr.ndmp_addr_u.tcp_addr.port);

			Debug(DBG_CAT_DATA|DBG_FOC_FLOW,
				  "ndmpdDataConnect: addr:%s port:%u\n",
				  inet_ntoa(sin.sin_addr),
				  (u_long)sin.sin_port);

			if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0)
			{
				Error(LOG_ERR, "ndmpdDataConnect: socket error: %s\n",
					  strerror(errno));
				reply.error = NDMP_CONNECT_ERR;
				break;
			}

			if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) < 0)
			{
				Error(LOG_ERR, "ndmpdDataConnect: connect error: %s\n",
					  strerror(errno));
				close(sock);
				reply.error = NDMP_CONNECT_ERR;
				break;
			}

			/*
			 * Set the O_NDELAY flag on the socket to prevent
			 * reads and writes from blocking.
			 */
			if ((flags = fcntl(sock, F_GETFL)) < 0)
			{
				Error(LOG_ERR, "ndmpdDataConnect: fcntl(F_GETFL) error: %s.\n",
					  strerror(errno));
				close(sock);
				reply.error = NDMP_CONNECT_ERR;
				break;
			}
		
			if ((flags = fcntl(sock,
							   F_SETFL, flags|O_NDELAY|O_NONBLOCK)) < 0)
			{
				Error(LOG_ERR, "ndmpdDataConnect: fcntl(F_SETFL) error: %s.\n",
					  strerror(errno));
				close(sock);
				reply.error = NDMP_CONNECT_ERR;
				break;
			}

			session->data.sock = sock;
			break;
		}

		case NDMP_ADDR_LOCAL:
		{
			/* Verify that the mover is listening for a local connection */
			if (session->mover.state != NDMP_MOVER_STATE_LISTEN ||
				session->mover.listenSock != -1)
			{
				Error(LOG_ERR,
					  "ndmpdDataConnect: mover is not in local listen state.\n");
				reply.error = NDMP_ILLEGAL_STATE_ERR;
				break;
			}

			session->mover.state = NDMP_MOVER_STATE_ACTIVE;
			break;
		}

		default:
		{
			Error(LOG_ERR,
				  "ndmpdDataConnect: invalid address type.\n");
			reply.error = NDMP_ILLEGAL_ARGS_ERR;
			break;
		}
	}
	
	if (reply.error == NDMP_NO_ERR)
	{
		session->data.state = NDMP_DATA_STATE_CONNECTED;
		session->data.moverAddr = request->addr;
	}
	
	if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
		Error(LOG_ERR,
			  "ndmpdDataConnect: error sending ndmp_data_connect_reply.\n");
}




/**** ndmpd global internal functions ************************************/

/*
 * ndmpdDataInit
 *   Initializes data specific session variables.
 *
 * Parameters:
 *   session (input) - session pointer.
 *
 * Returns:
 *   void
 */
void
ndmpdDataInit(NdmpdSession*		session)
{
	session->data.operation             = NDMP_DATA_OP_NOACTION;
	session->data.state                 = NDMP_DATA_STATE_IDLE;
	session->data.haltReason            = NDMP_DATA_HALT_NA;
	session->data.abortFlag             = FALSE;
	session->data.env                   = 0;
	session->data.envLen                = 0;
	session->data.nlist                 = 0;
	session->data.nlistLen              = 0;
	session->data.moverAddr.addr_type   = NDMP_ADDR_LOCAL;
	session->data.sock                  = -1;
	session->data.listenSock            = -1;
	session->data.readOffset            = 0;
	session->data.readLength            = 0;
	session->data.bytesLeftToRead       = 0;
	session->data.position              = 0;
	session->data.discardLength         = 0;
	session->data.module.stats.bytesProcessed    = 0;
	session->data.module.stats.estBytesRemaining = 0;
	session->data.module.stats.estTimeRemaining  = 0;
}


/*
 * ndmpdDataCleanup
 *   Releases resources allocated during a data operation.
 *
 * Parameters:
 *   session (input) - session pointer.
 *
 * Returns:
 *   void
 */
void
ndmpdDataCleanup(NdmpdSession*	session)
{
	if (session->data.moverAddr.addr_type == NDMP_ADDR_TCP)
	{
		if (session->data.sock != -1)
		{
			(void)ndmpdRemoveFileHandler(session, session->data.sock);
			(void)close(session->data.sock);
			session->data.sock = -1;
		}
		
		if (session->data.listenSock != -1)
		{
			(void)ndmpdRemoveFileHandler(session, session->data.listenSock);
			(void)close(session->data.listenSock);
			session->data.listenSock = -1;
		}
	}
	
	ndmpdFreeEnv(session);
	ndmpdFreeNlist(session);
}


/*
 * ndmpdDataError
 *   This function is called when a data error
 *   has been detected. A notify message is sent to the client and the
 *   data server is placed into the halted state.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   reason  (input) - halt reason.
 *
 * Returns:
 *   void
 */
void
ndmpdDataError(NdmpdSession*			session,
			   ndmp_data_halt_reason	reason)
{
	ndmp_notify_data_halted_request	req;
	
	if (session->data.state == NDMP_DATA_STATE_IDLE ||
		session->data.state == NDMP_DATA_STATE_HALTED)
		return;
	
	if (session->data.operation == NDMP_DATA_OP_BACKUP)
	{
		/*
		 * Send/discard any buffered file history data.
		 */
		ndmpdFileHistoryCleanup(session,
								(reason == NDMP_DATA_HALT_SUCCESSFUL ? TRUE : FALSE));

		/*
		 * If mover local and successful backup, write any
		 * remaining buffered data to tape.
		 */
		if (session->data.moverAddr.addr_type == NDMP_ADDR_LOCAL &&
			reason == NDMP_DATA_HALT_SUCCESSFUL)
		{
			(void)ndmpdLocalWrite(session, 0, 0);
		}
	}
	
	session->data.state      = NDMP_DATA_STATE_HALTED;
	session->data.haltReason = reason;
	
	req.reason      = session->data.haltReason;
	req.text_reason = "";
	
	if (ndmpSendRequest(session->connection, NDMP_NOTIFY_DATA_HALTED,
						NDMP_NO_ERR,
						(void *)&req, 0) < 0)
		Error(LOG_ERR,
			  "ndmpdApiDone: error sending notify_data_halted request.\n");
	
	if (session->data.moverAddr.addr_type == NDMP_ADDR_TCP)
	{
		if (session->data.sock != -1)
		{
			(void)ndmpdRemoveFileHandler(session,
										 session->data.sock);
			(void)close(session->data.sock);
			session->data.sock = -1;
		}
		if (session->data.listenSock != -1)
		{
			(void)ndmpdRemoveFileHandler(session,
										 session->data.listenSock);
			(void)close(session->data.listenSock);
			session->data.listenSock = -1;
		}
	}
	else
		ndmpdMoverError(session, NDMP_MOVER_HALT_CONNECT_CLOSED);
}


/*
 * ndmpdRemoteWrite
 *   Writes data to the remote mover.
 *
 * Parameters:
 *   session    (input) - session pointer.
 *   data       (input) - data to be written.
 *   length     (input) - data length.
 *
 * Returns:
 *   0 - data successfully written.
 *  -1 - error.
 */
int
ndmpdRemoteWrite(NdmpdSession*	session,
				 char*			data,
				 u_long			length)
{
	ssize_t		n;
	u_long		count = 0;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdRemoteWrite: length:%u\n", length);

	while (count < length)
	{
		/*
		 * Wait for the connection to become ready for more data.
		 * We can't afford to block on the write since NDMP requests
		 * also need to be processed. Blocking could lead to deadlocks
		 * depending on the sequence the NDMP client uses to send
		 * mover and data server requests.
		 */
		if (ndmpdAddFileHandler(session, (void*)session,
								session->data.sock,
								NDMPD_SELECT_MODE_WRITE,
								HC_MOVER,
								dataReady) < 0)
			return(-1);
			
		session->data.ioReady = FALSE;
			
		for (;;)
		{
			if (ndmpdSelect(session, TRUE, HC_CLIENT|HC_MOVER) < 0)
				return(-1);

			if (session->eof == TRUE ||
				session->data.abortFlag == TRUE)
				return(-1);

			if (session->data.ioReady == TRUE)
				break;
		}
			
		if ((n = write(session->data.sock,
					   &data[count], length - count)) < 0)
		{
			Error(LOG_ERR,
				  "ndmpdWrite: write error: %s.\n",
				  strerror(errno));
			return(-1);
		}

		count += n;
	}

	return(0);
}
	

/*
 * ndmpdRemoteRead
 *   Reads data from the remote mover.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   data    (input) - data to be written.
 *   length  (input) - data length.
 *
 * Returns:
 *   0 - data successfully read.
 *  -1 - error.
 */
int
ndmpdRemoteRead(NdmpdSession*	session,
				char*			data,
				u_long			length)
{
	u_long			count = 0;
	ssize_t			n;
	u_long			len;
		
	Debug(DBG_CAT_DATA|DBG_FOC_FLOW,
		  "ndmpdRemoteRead: length:%u\n", length);

	while (count < length)
	{
		len = length - count;

		/*
		 * If the end of the seek window has been reached then
		 * send an ndmp_read request to the client.
		 * The NDMP client will then send a mover_data_read request to
		 * the remote mover and the mover will send more data.
		 * This condition can occur if the module attempts to read past
		 * a seek window set via a prior call to ndmpdSeek() or
		 * the module has not issued a seek. If no seek was issued then
		 * pretend that a seek was issued to read the entire tape.
		 */
		if (session->data.bytesLeftToRead == 0)
		{
			ndmp_notify_data_read_request	request;

			/* ndmpdSeek() never called? */
			if (session->data.readLength == 0)
			{
				session->data.bytesLeftToRead  = ~0LL;
				session->data.readOffset       = 0LL;
				session->data.readLength       = ~0LL;
			}
			else
			{
				session->data.bytesLeftToRead  = len;
				session->data.readOffset       = session->data.position;
				session->data.readLength       = len;
			}
			
			request.offset = longLongToQuad(session->data.readOffset);
			request.length = longLongToQuad(session->data.readLength);
		
			if (ndmpSendRequest(session->connection, NDMP_NOTIFY_DATA_READ,
								NDMP_NO_ERR, (void *)&request, 0) < 0)
			{
				Error(LOG_ERR,
					  "ndmpdRemoteRead: error sending notify_data_read request.\n");
				return(-1);
			}
		}
			
		/*
		 * Wait for data to be available to be read.
		 * We can't afford to block on the read since NDMP requests
		 * also need to be processed. Blocking could lead to deadlocks
		 * depending on the sequence the NDMP client uses to send
		 * mover and data server requests.
		 */
		if (ndmpdAddFileHandler(session, (void*)session,
								session->data.sock,
								NDMPD_SELECT_MODE_READ,
								HC_MOVER,
								dataReady) < 0)
			return(-1);
			
		session->data.ioReady = FALSE;
			
		for (;;)
		{
			if (ndmpdSelect(session, TRUE, HC_CLIENT|HC_MOVER) < 0)
				return(-1);

			if (session->eof == TRUE ||
				session->data.abortFlag == TRUE)
				return(-1);

			if (session->data.ioReady == TRUE)
				break;
		}
			
		/*
		 * If the module called ndmpdSeek() prior to reading all of the
		 * data that the remote mover was requested to send, then the
		 * excess data from the seek has to be discarded.
		 */
		if (session->data.discardLength != 0)
		{
			n = discardData(session, (u_long)session->data.discardLength);
			if (n < 0)
				return(-1);
			session->data.discardLength -= n;
			continue;
		}
		
		/*
		 * Don't attempt to read more data than the remote is sending.
		 */
		if (len > session->data.bytesLeftToRead)
			len = session->data.bytesLeftToRead;
			
		if ((n = read(session->data.sock,
					  &data[count], len)) < 0)
		{
			Error(LOG_ERR,
				  "ndmpdRemoteRead: read error: %s.\n",
				  strerror(errno));
			return(-1);
		}

		/* read returns 0 if the connection was closed */
		if (n == 0)
			return(-1);
		
		count += n;
		session->data.bytesLeftToRead -= n;
		session->data.position += n;
	}

	return(0);
}


/**** static functions *********************************************/

/*
 * startBackup
 *
 * Parameters:
 *   session   (input) - session pointer.
 *   bu_type   (input) - backup type.
 *   env_val   (input) - environment variable array.