mover.c

网络数据管理协议的开发

		Error(LOG_ERR,
			  "ndmpdMoverRead: tape device is not open.\n");
		reply.error = NDMP_DEV_NOT_OPEN_ERR;
		if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
			Error(LOG_ERR,
				  "ndmpdMoverRead: error sending ndmp_mover_read_reply.\n");

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

	err = ndmpdMoverSeek(session,
						 quadToLongLong(request->offset),
						 quadToLongLong(request->length));
	if (err < 0)
	{
		ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);
		return;
	}
	
	/*
	 * Just return if we are waiting for the NDMP client to
	 * complete the seek.
	 */
	if (err == 1)
		return;
	
	/*
	 * Setup a handler function that will be called when
	 * data can be written to the data connection without blocking.
	 */
	if (ndmpdAddFileHandler(session, (void*)session,
							session->mover.sock,
							NDMPD_SELECT_MODE_WRITE,
							HC_MOVER,
							moverDataWrite) < 0)
	{
		ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);
		return;
	}
}


/*
 * ndmpdMoverClose
 *
 * This handler handles ndmp_mover_close_requests.
 *
 * Parameters:
 *   connection (input) - connection handle.
 *   body       (input) - request message body.
 *
 * Returns:
 *   void
 */
void
ndmpdMoverClose(NdmpConnection	connection,
				void*			body __attribute__ ((unused)))
{
	ndmp_mover_close_reply		reply;
	NdmpdSession*				session = ndmpGetClientData(connection);
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdMoverClose.\n");

	memset((void*)&reply, 0, sizeof(reply));
	
	if (session->mover.state != NDMP_MOVER_STATE_PAUSED)
	{
		Error(LOG_ERR,
			  "ndmpdMoverClose: invalid state.\n");

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

	reply.error = NDMP_NO_ERR;
	if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
		Error(LOG_ERR,
			  "ndmpdMoverClose: error sending mover_close reply.\n");

	ndmpdMoverError(session, NDMP_MOVER_HALT_CONNECT_CLOSED);
}


/*
 * ndmpdMoverSetRecordSize
 *
 * This handler handles ndmp_mover_set_record_size_requests.
 *
 * Parameters:
 *   connection (input) - connection handle.
 *   body       (input) - request message body.
 *
 * Returns:
 *   void
 */
void
ndmpdMoverSetRecordSize(NdmpConnection	connection,
						void*			body)
{
	ndmp_mover_set_record_size_request*	request	=
		(ndmp_mover_set_record_size_request*)body;
	ndmp_mover_set_record_size_reply	reply;
	NdmpdSession*					session = ndmpGetClientData(connection);
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdMoverSetRecordSize: len:%llu.\n",
		  request->len);

	memset((void*)&reply, 0, sizeof(reply));
	
	session->mover.recordSize = request->len;

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


/*
 * ndmpdMoverConnect
 *   Request handler. Connects the mover to either a local
 *   or remote data server.
 *
 * Parameters:
 *   connection (input) - connection handle.
 *   body       (input) - request message body.
 *
 * Returns:
 *   void
 */
void
ndmpdMoverConnect(NdmpConnection	connection,
				  void*				body)
{
	ndmp_mover_connect_request*	request 
		= (ndmp_mover_connect_request*)body;
	ndmp_mover_connect_reply	reply;
	NdmpdSession*				session = ndmpGetClientData(connection);
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdMoverConnect: mode:%s type:%s.\n",
		  request->mode == NDMP_MOVER_MODE_READ ? "READ" : "WRITE",
		  request->addr.addr_type == NDMP_ADDR_LOCAL ? "LOCAL" :
		  request->addr.addr_type == NDMP_ADDR_TCP ? "TCP" :
		  request->addr.addr_type == NDMP_ADDR_FC ? "FC" :
		  request->addr.addr_type == NDMP_ADDR_IPC ? "IPC" :
		  "undefined");

	memset((void*)&reply, 0, sizeof(reply));
	
	if (session->mover.state != NDMP_MOVER_STATE_IDLE)
	{
		Error(LOG_ERR,
			  "ndmpdMoverConnect: 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,
				  "ndmpdMoverConnect: error sending ndmp_mover_connect_reply.\n");
		return;
	}

	if (session->tape.fd == -1)
	{
		Error(LOG_ERR,
			  "ndmpdMoverConnect: no tape device open.\n");

		reply.error = NDMP_DEV_NOT_OPEN_ERR;
		if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
			Error(LOG_ERR,
				  "ndmpdMoverConnect: error sending ndmp_mover_connect_reply.\n");
		return;
	}
	
	if (request->mode == NDMP_MOVER_MODE_READ &&
		session->tape.mode == NDMP_TAPE_READ_MODE)
	{
		Error(LOG_ERR,
			  "ndmpdMoverConnect: write protected device.\n");
		reply.error = NDMP_WRITE_PROTECT_ERR;
		if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
			Error(LOG_ERR,
				  "ndmpdMoverConnect: error sending ndmp_mover_connect_reply.\n");
		return;
	}

	reply.error = NDMP_NO_ERR;

	switch (request->addr.addr_type)
	{
		case NDMP_ADDR_LOCAL:
		{
			/* Verify that the data server is listening for a local connection */
			if (session->data.state != NDMP_DATA_STATE_LISTEN ||
				session->data.listenSock != -1)
			{
				Error(LOG_ERR,
					  "ndmpdMoverConnect: data server is not in local listen state.\n");
				reply.error = NDMP_ILLEGAL_STATE_ERR;
				break;
			}

			session->data.state = NDMP_DATA_STATE_CONNECTED;
			break;
		}

		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_MOVER|DBG_FOC_FLOW,
				  "ndmpdMoverConnect: 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, "ndmpdMoverConnect: socket error: %s\n",
					  strerror(errno));
				reply.error = NDMP_CONNECT_ERR;
				break;
			}

			if (connect(sock, (struct sockaddr *)&sin, sizeof(sin)) < 0)
			{
				Error(LOG_ERR, "ndmpdMoverConnect: 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, "ndmpdMoverConnect: 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, "ndmpdMoverConnect: fcntl(F_SETFL) error: %s.\n",
					  strerror(errno));
				close(sock);
				reply.error = NDMP_CONNECT_ERR;
				break;
			}

			if (request->mode == NDMP_MOVER_MODE_READ)
			{
				if (ndmpdAddFileHandler(session, (void*)session,
										sock,
										NDMPD_SELECT_MODE_READ,
										HC_MOVER,
										moverDataRead) < 0)
				{
					close(sock);
					reply.error = NDMP_CONNECT_ERR;
					break;
				}
			}
	
			session->mover.sock = sock;
			break;
		}

		default:
		{
			Error(LOG_ERR,
				  "ndmpdMoverConnect: invalid address type.\n");
			reply.error = NDMP_ILLEGAL_ARGS_ERR;
			break;
		}
	}
	
	if (reply.error == NDMP_NO_ERR)
	{
		session->mover.state = NDMP_MOVER_STATE_ACTIVE;
		session->mover.mode  = request->mode;
	}
	
	if (ndmpSendReply(connection, NDMP_NO_ERR, (void *)&reply) < 0)
		Error(LOG_ERR,
			  "ndmpdMoverConnect: error sending ndmp_mover_listen_reply.\n");
}


/*
 * ndmpdLocalWrite
 *   Buffers and writes data to the tape device.
 *   A full tape record is buffered before being written.
 *
 * Parameters:
 *   session    (input) - session pointer.
 *   data       (input) - data to be written.
 *   length     (input) - data length.
 *
 * Returns:
 *   0 - data successfully written.
 *  -1 - error.
 */
int
ndmpdLocalWrite(NdmpdSession*	session,
				char*			data,
				u_long			length)
{
	u_long			count = 0;
	ssize_t			n;
	u_long			len;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdLocalWrite: length:%u\n", length);

	if (session->mover.state == NDMP_MOVER_STATE_IDLE ||
		session->mover.state == NDMP_MOVER_STATE_LISTEN ||
		session->mover.state == NDMP_MOVER_STATE_HALTED)
	{
		Error(LOG_ERR,
			  "ndmpdLocalWrite: invalid mover state to write data.\n");
		return(-1);
	}
	
	/*
	 * A length of 0 indicates that any buffered data should be
	 * flushed to tape.
	 */
	if (length == 0)
	{
		if (session->mover.wIndex == 0)
			return(0);

		memset((void*)&session->mover.buf[session->mover.wIndex],
			   0, session->mover.recordSize - session->mover.wIndex);

		n = moverTapeWrite(session,
					  session->mover.buf,
					  session->mover.recordSize);
		if (n <= 0)
		{
			ndmpdMoverError(session, (n == 0 ? NDMP_MOVER_HALT_ABORTED
									  : NDMP_MOVER_HALT_INTERNAL_ERROR));
			return(-1);
		}
		
		session->mover.dataWritten += n;
		session->mover.recordNum++;
		session->mover.wIndex = 0;
		return(0);
	}
	
	/* Break the data into records. */
	while (count < length)
	{
		/*
		 * Determine if data needs to be buffered or
		 * can be written directly from user supplied location.
		 * We can fast path the write if there is no pending
		 * buffered data and there is at least a full records worth
		 * of data to be written.
		 */
		if (session->mover.wIndex == 0 &&
			length - count >= session->mover.recordSize)
		{
			n = moverTapeWrite(session, &data[count],
						  session->mover.recordSize);
			if (n <= 0)
			{
				ndmpdMoverError(session, (n == 0 ? NDMP_MOVER_HALT_ABORTED
										  : NDMP_MOVER_HALT_INTERNAL_ERROR));
				return(-1);
			}
			
			session->mover.dataWritten += n;
			session->mover.recordNum++;
			count += n;
			continue;
		}

		/* Buffer the data */
		len = length - count;
		if (len > session->mover.recordSize - session->mover.wIndex)
			len = session->mover.recordSize - session->mover.wIndex;
		
		memcpy(&session->mover.buf[session->mover.wIndex],
			   &data[count], len);
		session->mover.wIndex += len;
		count += len;

		/* Write the buffer if its full */
		if (session->mover.wIndex == session->mover.recordSize)
		{
			n = moverTapeWrite(session,
						  session->mover.buf,
						  session->mover.recordSize);
			if (n < 0)
			{
				ndmpdMoverError(session, (n == 0 ? NDMP_MOVER_HALT_ABORTED
										  : NDMP_MOVER_HALT_INTERNAL_ERROR));
				return(-1);
			}
			
			session->mover.dataWritten += n;
			session->mover.recordNum++;
			session->mover.wIndex = 0;
		}
	}

	return(0);
}


/*
 * ndmpdLocalRead
 *   Reads data from the local tape device.
 *   Full tape records are read and buffered.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   data    (input) - location to store data.
 *   length  (input) - data length.
 *
 * Returns:
 *   0 - data successfully read.
 *  -1 - error.
 */
int
ndmpdLocalRead(NdmpdSession*	session,
			   char*			data,
			   u_long			length)
{
	u_long			count = 0;
	ssize_t			n;
	u_long			len;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "ndmpdLocalRead: length:%u\n", length);

	if (session->mover.state == NDMP_MOVER_STATE_IDLE ||
		session->mover.state == NDMP_MOVER_STATE_LISTEN ||
		session->mover.state == NDMP_MOVER_STATE_HALTED)
	{
		Error(LOG_ERR,
			  "ndmpdLocalRead: invalid mover state to read data.\n");
		return(-1);
	}
	
	/*
	 * Automatically increase the seek window if necessary.
	 * This is needed in the event 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 (length > session->mover.bytesLeftToRead)
	{
		/* ndmpdSeek() never called? */
		if (session->data.readLength == 0)
		{
			session->mover.bytesLeftToRead = ~0LL;
			session->data.readOffset       = 0LL;
			session->data.readLength       = ~0LL;
		}
		else
		{
			session->mover.bytesLeftToRead = length;
			session->data.readOffset       = session->mover.position;
			session->data.readLength       = length;
		}
	}

	/*
	 * Read as many records as necessary to satisfy the request.
	 */
	while (count < length)
	{
		/*
		 * If the end of the mover window has been reached,
		 * then notify the client that a new data window is needed.
		 */
		if (session->mover.position >=
			session->mover.windowOffset + session->mover.windowLength)
		{
			ndmp_notify_mover_paused_request	pauseRequest;

			session->mover.state        = NDMP_MOVER_STATE_PAUSED;
			session->mover.pauseReason  = NDMP_MOVER_PAUSE_SEEK;
			pauseRequest.reason         = NDMP_MOVER_PAUSE_SEEK;
			pauseRequest.seek_position  = longLongToQuad(session->mover.position);

			if (ndmpSendRequest(session->connection, NDMP_NOTIFY_MOVER_PAUSED,
								NDMP_NO_ERR, (void *)&pauseRequest, 0) < 0)