mover.c

网络数据管理协议的开发

	}
		
	if ((flags = fcntl(session->mover.sock,
					   F_SETFL, flags|O_NDELAY|O_NONBLOCK)) < 0)
	{
		Error(LOG_ERR, "moverAcceptConnection: fcntl(F_SETFL) error: %s.\n",
			  strerror(errno));
		ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);
		return;
	}

	if (session->mover.mode == NDMP_MOVER_MODE_READ)
	{
		if (ndmpdAddFileHandler(session, (void*)session,
								session->mover.sock,
								NDMPD_SELECT_MODE_READ,
								HC_MOVER,
								moverDataRead) < 0)
		{
			ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);
			return;
		}
	}
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "moverAcceptConnection: Received connection from %s\n",
		  inet_ntoa(from.sin_addr));

	session->mover.state = NDMP_MOVER_STATE_ACTIVE;
}


/*
 * moverDataRead
 *   Reads backup data from the data connection and writes the
 *   received data to the tape device.
 *
 * Parameters:
 *   cookie  (input) - session pointer.
 *   fd      (input) - file descriptor.
 *   mode    (input) - select mode.
 *
 * Returns:
 *   void.
 */
static void
moverDataRead(void*		cookie,
			  int		fd,
			  u_long	mode __attribute__ ((unused)))
{
	NdmpdSession*	session = (NdmpdSession*)cookie;
	int				n;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "moverDataRead.\n");

	n = read(fd,
			 &session->mover.buf[session->mover.wIndex],
			 session->mover.recordSize - session->mover.wIndex);

	/*
	 * Since this function is only called when select believes data
	 * is available to be read, a return of zero indicates the
	 * connection has been closed.
	 */
	if (n <= 0)
	{
		u_long	index;
		
		if (n < 0 && errno == EWOULDBLOCK)
			return;
			
		/* Save the index since moverTapeFlush resets it. */
		index = session->mover.wIndex;

		/* Flush any buffered data to tape. */
		if (moverTapeFlush(session) >= 0)
		{
			session->mover.dataWritten += index;
			session->mover.recordNum++;
		}
		
		if (n == 0)
			ndmpdMoverError(session, NDMP_MOVER_HALT_CONNECT_CLOSED);
		else
			ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);

		return;
	}

	session->mover.wIndex += n;

	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;
		}
		
		session->mover.wIndex = 0;
		session->mover.dataWritten += n;
		session->mover.recordNum++;
	}
}


/*
 * moverDataWrite
 *   Reads backup data from the tape device and writes the
 *   data to the data connection.
 *   This function is called by ndmpdSelect when the data connection
 *   is ready for more data to be written.
 *
 * Parameters:
 *   cookie  (input) - session pointer.
 *   fd      (input) - file descriptor.
 *   mode    (input) - select mode.
 *
 * Returns:
 *   void.
 */
static void
moverDataWrite(void*		cookie,
			   int			fd,
			   u_long		mode __attribute__ ((unused)))
{
	NdmpdSession*	session = (NdmpdSession*)cookie;
	int				n;
	u_long			len;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "moverDataWrite: rIndex:%u wIndex:%u bytesLeftToRead:%llu.\n",
		  session->mover.rIndex, session->mover.wIndex,
		  session->mover.bytesLeftToRead);

	/*
	 * Read more data into the tape buffer if the buffer is empty.
	 */
	if (session->mover.wIndex == 0)
	{
		/*
		 * If the end of the mover window has been reached,
		 * then notify the client that a seek is needed.
		 * Remove the file handler to prevent this function from
		 * being called. The handler will be reinstalled in
		 * ndmpdMoverContinue.
		 */
		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);

			(void)ndmpdRemoveFileHandler(session, fd);

			if (ndmpSendRequest(session->connection, NDMP_NOTIFY_MOVER_PAUSED,
								NDMP_NO_ERR, (void *)&pauseRequest, 0) < 0)
			{
				Error(LOG_ERR,
					  "moverDataWrite: error sending notify_mover_paused request.\n");
				ndmpdMoverError(session, NDMP_MOVER_HALT_INTERNAL_ERROR);
			}
			return;
		}

		n = moverTapeRead(session, session->mover.buf);

		Debug(DBG_CAT_MOVER|DBG_FOC_DETAIL,
			  "moverDataWrite: read %u bytes from tape.\n", n);

		if (n <= 0)
		{
			ndmpdMoverError(session, (n == 0 ? NDMP_MOVER_HALT_ABORTED
									  : NDMP_MOVER_HALT_INTERNAL_ERROR));
			return;
		}
		
		/*
		 * Discard data if the current data stream position is
		 * prior to the seek position. This is necessary if a seek
		 * request set the seek pointer to a position that is not a
		 * record boundary. The seek request handler can only position
		 * to the start of a record.
		 */
		if (session->mover.position < session->mover.seekPosition)
		{
			session->mover.rIndex = session->mover.seekPosition -
				session->mover.position;
			session->mover.position = session->mover.seekPosition;
		}
		
		session->mover.wIndex = n;
	}
	
	/*
	 * The limit on the total amount of data to be sent can be
	 * dictated by either the end of the mover window or the end of the
	 * seek window. 
	 * First determine which window applies and then determine if the
	 * send length needs to be less than a full record to avoid
	 * exceeding the window.
	 */
	if (session->mover.position + session->mover.bytesLeftToRead >
		session->mover.windowOffset + session->mover.windowLength)
		len = session->mover.windowOffset + session->mover.windowLength -
			session->mover.position;
	else
		len = session->mover.bytesLeftToRead;

	/*
	 * Now limit the length to the amount of data in the buffer.
	 */
	if (len > session->mover.wIndex - session->mover.rIndex)
		len = session->mover.wIndex - session->mover.rIndex;
	
	/*
	 * Write the data to the data connection.
	 */
	n = write(session->mover.sock,
			  &session->mover.buf[session->mover.rIndex],
			  len);

	if (n < 0)
	{
		if (errno == EWOULDBLOCK)
			return;
			
		ndmpdMoverError(session, NDMP_MOVER_HALT_CONNECT_CLOSED);
		return;
	}
	
	Debug(DBG_CAT_MOVER|DBG_FOC_DETAIL,
		  "moverDataWrite: wrote %u of %u bytes to data connection; position:%llu rIndex:%lu.\n",
		  n, len, session->mover.position, session->mover.rIndex);

	session->mover.rIndex          += n;
	session->mover.position        += n;
	session->mover.bytesLeftToRead -= n;

	/*
	 * If all data in the buffer has been written,
	 * zero the buffer indices. The next call to this function
	 * will read more data from the tape device into the buffer.
	 */
	if (session->mover.rIndex == session->mover.wIndex)
	{
		session->mover.rIndex = 0;
		session->mover.wIndex = 0;
	}

	/*
	 * If the read limit has been reached,
	 * then remove the file handler to prevent this
	 * function from getting called. The next mover_read request
	 * will reinstall the handler.
	 */
	if (session->mover.bytesLeftToRead == 0)
		(void)ndmpdRemoveFileHandler(session, fd);
}


/*
 * moverTapeWrite
 *   Writes a data record to tape. Detects and handles EOT conditions.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   data    (input) - data to be written.
 *   length  (input) - length of data to be written.
 *
 * Returns:
 *    0 - operation aborted by client.
 *   -1 - error.
 *   otherwise - number of bytes written.
 */
static int
moverTapeWrite(NdmpdSession*	session,
			   char*			data,
			   ssize_t			length)
{
	ssize_t		n;
	int			err;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "moverTapeWrite: length:%u\n", length);

	for (;;)
	{
		if (session->tape.recordCount == DEBUG_TAPE_SIZE)
			n = 0;
		else
		{
			n = write(session->tape.fd, data, length);
			if (n < 0)
			{
				Error(LOG_ERR,
					  "moverTapeWrite: tape write error: %s.\n",
					  strerror(errno));
				return(-1);
			}
		}
		
		if (n == 0)
		{
			/*
			 * End of media reached.
			 * Notify client and wait for the client to
			 * either abort the operation or continue the
			 * operation after changing the tape.
			 */
			ndmpdApiLog((void*)session, NDMP_LOG_NORMAL, 0,
						"End of tape reached. Load next tape.\n");
			
			err = moverChangeTape(session);
			
			/* Operation aborted or connection terminated? */
			if (err < 0)
				return(-1);

			/* Retry the write to the new tape. */
			continue;
		}

		session->tape.recordCount++;
		return(n);
	}
}


/*
 * tapeFLush
 *   Writes all remaining buffered data to tape. A partial record is
 *   padded out to a full record with zeros.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   data    (input) - data to be written.
 *   length  (input) - length of data to be written.
 *
 * Returns:
 *   -1 - error.
 *   otherwise - number of bytes written.
 */
static int
moverTapeFlush(NdmpdSession*	session)
{
	int	n;
	
	Debug(DBG_CAT_MOVER|DBG_FOC_FLOW,
		  "moverTapeFlush.\n");

	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)
	{
		Error(LOG_ERR,
			  "moverTapeFlush: write error: %s.\n",
			  strerror(errno));
		return(-1);
	}
	
	session->mover.wIndex = 0;
	return(0);
}


/*
 * moverTapeRead
 *   Reads a data record from tape. Detects and handles EOT conditions.
 *
 * Parameters:
 *   session (input) - session pointer.
 *   data    (input) - location to read data to.
 *
 * Returns:
 *    0 - operation aborted.
 *   -1 - tape read error.
 *   otherwise - number of bytes read.
 */
static int
moverTapeRead(NdmpdSession*	session,
		 char*				data)
{
	ssize_t		n;
	int			err;
	
	for (;;)
	{
		if (session->tape.recordCount == DEBUG_TAPE_SIZE)
			n = 0;
		else
		{
			n = read(session->tape.fd, data, session->mover.recordSize);
			if (n < 0)
			{
				Error(LOG_ERR,
					  "moverTapeRead: tape read error: %s.\n",
					  strerror(errno));
				return(-1);
			}
		}
		
		if (n == 0)
		{
			/*
			 * End of media reached.
			 * Notify client and wait for the client to
			 * either abort the data operation or continue the
			 * operation after changing the tape.
			 */
			ndmpdApiLog((void*)session, NDMP_LOG_NORMAL, 0,
						"End of tape reached. Load next tape.\n");
			
			err = moverChangeTape(session);
			
			/* Operation aborted or connection terminated? */
			if (err < 0)
				return(0);

			/* Retry the read from the new tape. */
			continue;
		}

		session->mover.recordNum++;
		session->tape.recordCount++;
		return(n);
	}
}


/*
 * moverChangeTape
 *   Send an ndmp_notify_mover_paused request to the
 *   NDMP client to inform the client that a tape volume change is required.
 *   Process messages until the data/mover operation is either aborted
 *   or continued.
 *
 * Parameters:
 *   clientData (input) - session pointer.
 *
 * Returns:
 *   0 - operation has been continued.
 *  -1 - operation has been aborted.
 */
static int
moverChangeTape(NdmpdSession*		session)
{
	ndmp_notify_mover_paused_request	request;

	session->mover.state = NDMP_MOVER_STATE_PAUSED;
	
	if (session->mover.mode == NDMP_MOVER_MODE_READ)
		session->mover.pauseReason = NDMP_MOVER_PAUSE_EOM;
	else
		session->mover.pauseReason = NDMP_MOVER_PAUSE_EOF;
	
	request.reason = session->mover.pauseReason;
	request.seek_position  = longLongToQuad(0LL);

	if (ndmpSendRequest(session->connection, NDMP_NOTIFY_MOVER_PAUSED,
						NDMP_NO_ERR, (void *)&request, 0) < 0)
	{
		Error(LOG_ERR,
			  "moverChangeTape: error sending notify_mover_paused_request.\n");
		return(-1);
	}

	/*
	 * Process messages until the state is changed by
	 * an abort, continue, or close request .
	 */
	for (;;)
	{
		if (ndmpdSelect(session, TRUE, HC_CLIENT) < 0)
			return(-1);

		if (session->eof == TRUE)
			return(-1);
		
		switch (session->mover.state)
		{
			case NDMP_MOVER_STATE_ACTIVE:
				session->tape.recordCount = 0;
				return(0);
				
			case NDMP_MOVER_STATE_PAUSED:
				continue;
					
			default:
				return(-1);
		}
	}
}