ftp_client_pipe.cc

实现了poll/epoll/devpoll等C++封装

			/* Caller will call stop() */
			return connecterr;
		} else {
			m_cfd_connecting = false;
			DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: Connect cfd %d succeeded\n", m_cfd));
		}
	}
	if (revents & POLLIN) {
		/* Read it */
		int err = m_ibuf.fillFrom(m_cfd);
		if (err && (err != EWOULDBLOCK)) {
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: fillFrom returns %d?\n", err));
			return err;
		}
		// You must tell poller about EWOULDBLOCK -- it has no other way to
		// know that a socket is no longer ready for I/O!
		if (err == EWOULDBLOCK)
			m_poller->clearReadiness(m_cfd, POLLIN);
	}
	//DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd(fd %d, %d): before loop: m_proto.isInputReady() %d, m_proto.isOutputReady() %d\n", fd, (int) revents, m_proto.isInputReady(), m_proto.isOutputReady())); 
	/* Process all lines in m_ibuf */
	int puterr = 0;
	int geterr = 0;
	while ((m_proto.isInputReady() && (geterr != EWOULDBLOCK))
	    || (m_proto.isOutputReady() && (puterr != EWOULDBLOCK))) {
		//DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd(%d): m_proto.isInputReady() %d, geterr %d, m_proto.isOutputReady() %d, puterr %d\n", revents, m_proto.isInputReady(), geterr, m_proto.isOutputReady(), puterr)); 
		if (!m_iline_full) {
			geterr = m_ibuf.readline(m_iline, sizeof(m_iline));
			if (!geterr) {
				m_iline_full = true;
				DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd fd %d got '%s'\n", m_cfd, m_iline));
			} else if (geterr != EWOULDBLOCK)
				return geterr;
		}
		if (m_iline_full) {
			if (m_state == IDLE) {
				/* cancel timeout*/
				m_sked->delClient(this);
			}

			/* A line is ready.  Process it. */
			int err = m_proto.giveInput(m_iline);
			if (!err) {
				m_iline_full = false;
			} else if (err != EWOULDBLOCK) {
				EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: m_proto.giveInput returns %d\n",err));
				return err;
			}
			status = m_proto.getStatus(NULL, 0);
			if (status) {
				/* If a command is finished, so are we. */
				return EALREADY;
			}
		}
		int len;
		const char *p;
		while ((p = m_proto.getOutput(&len)) != NULL) {
			DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd fd %d put '%s'\n", m_cfd, p));
			puterr = m_obuf.put(p, len);
			if (puterr == EWOULDBLOCK) {
				DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: put returns %d?\n", puterr));
				break;
			} else if (puterr) {
				EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: put returns %d?\n", puterr));
				return puterr;
			}
			m_proto.advanceOutput();
		}
	}
	// Try writing even if we don't know that we're writable --
	// since we didn't express interest in writing before,
	// this should save us one notification cycle.
	// 'Tis easier to ask forgiveness than permission.
	//if (revents & POLLOUT) {
		if (!m_cfd_connecting) {
			/* Write buffer to network. */
			int err = m_obuf.flushTo(m_cfd);
			if (err && (err != EWOULDBLOCK)) {
				EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: flushTo returns %d?\n", err));
				return err;
			}
			// You must tell poller about EWOULDBLOCK -- it has no other way to
			// know that a socket is no longer ready for I/O!
			if (err == EWOULDBLOCK)
				m_poller->clearReadiness(m_cfd, POLLOUT);
		}
		if (m_obuf.isEmpty() && m_proto.isQuit()) {
			DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: session quit, output done, please shut me down\n"));
			return EPIPE;
		}
	//}

	/* If it's time for us to open a data connection to a port on the
	 * server, do so
	 */
#define GETBYTE3(x) (((x) >> 24) & 255)
#define GETBYTE2(x) (((x) >> 16) & 255)
#define GETBYTE1(x) (((x) >> 8) & 255)
#define GETBYTE0(x) ((x) & 255)
	struct sockaddr_in address;
	if (m_proto.isPortReady(&address)) {
		int sock;

#ifdef USE_DPRINT
		int addr = ntohl(address.sin_addr.s_addr);
		int port = ntohs(address.sin_port);

		DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: isPortReady TRUE, address %d.%d.%d.%d:%d\n", 
			GETBYTE3(addr), 
			GETBYTE2(addr), 
			GETBYTE1(addr), 
			GETBYTE0(addr), 
			ntohs(port)));
#endif

		if ((sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
			xerr = errno;
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: connect failed, errno %d\n", xerr));
			/* Caller will call stop() */
			return xerr;
		}

		if (setNonblocking(sock) == -1) {
			xerr = errno;
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: setNonblocking failed, returning %d\n", xerr));
			/* Caller will call stop() */
			return xerr;
		}
		m_dfd_connecting = true;
		m_dfd = sock;
		int err = m_poller->add(m_dfd, this, POLLIN|POLLOUT);
		if (err) {
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: add failed\n"));
			return err;
		}
		if (ephemeral_connect(sock, &address, m_local_addr)) {
			if (errno != EINPROGRESS) {
				m_dfd_connecting = false;
				xerr = errno;
				EDPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: Connect fd %d failed, returning %d\n", sock, xerr));
				/* Caller will call stop() */
				return xerr;
			}
		} else {
			DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: Connect fd %d succeeded early\n", sock));
			m_dfd_connecting = false;
		}
		DPRINT(("ftp_client_pipe_t::notifyPollEvent_cfd: socket returns dfd %d\n", m_dfd));
	}

	return 0;
}

/*----------------------------------------------------------------------
 The operating system has told us that one of our file descriptors
 is ready for I/O.  Deal with it.
 fd and revents are from poll()'s output array or sigtimedwait();

 Returns 0 on success, Unix error code on failure.
 If this returns an error, call shutdown() to close this session.

 This is normally called by the app after a call to poll() or 
 sigtimedwait(), or internally by skedCallback().
----------------------------------------------------------------------*/
int ftp_client_pipe_t::notifyPollEvent(Poller::PollEvent *event)
{
	int xerr;		/* error to call ftpClientDone() with */
	int err;

	DPRINT(("ftp_client_pipe_t::notifyPollEvent(fd %d, %x): state %d\n", event->fd, event->revents, m_state));
	assert(event->fd != -1);

	if (event->fd == m_dfd)
		xerr = notifyPollEvent_dfd(event);
	else if (event->fd == m_cfd)
		xerr = notifyPollEvent_cfd(event);
	else {
		EDPRINT(("ftp_client_pipe_t::notifyPollEvent: fd %d != cfd %d or dfd %d!\n", event->fd, m_cfd, m_dfd));
		return EINVAL;
	}
	DPRINT(("ftp_client_pipe_t::notifyPollEvent: xerr %d\n", xerr));

	/* Update interest mask for Control */
	if (m_cfd != -1) {
		short cevents = 0;
		if (!m_ibuf.isFull())
			cevents |= POLLIN;
		if (!m_obuf.isEmpty())
			cevents |= POLLOUT;
		DPRINT(("ftp_client_pipe_t::notifyPollEvent: adding m_cfd %d, events %x\n", m_cfd, cevents));
		err = m_poller->setMask(m_cfd, cevents);
		if (err) {
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent: setMask failed\n"));
			exit(1);
		}
	}

	/* Update interest mask for Data */
	if (m_dfd != -1) {
		short devents = 0;
		/* If we're sleeping, don't ask for events if we already have them */
		if ((m_state == GETTING) || ((m_state == SLEEPING) && !m_dfd_events))
			devents = POLLIN;
		else if (m_state == PUTTING)
			devents = POLLOUT;
		DPRINT(("ftp_client_pipe_t::notifyPollEvent: adding m_dfd %d, events %x\n", m_dfd, devents));
		err = m_poller->setMask(m_dfd, devents);
		if (err) {
			EDPRINT(("ftp_client_pipe_t::notifyPollEvent: setMask failed\n"));
			exit(1);
		}
	}

	/* If no error, just return. */
	if (xerr == 0)
		return 0;
	// ok to have EALREADY; that's what is returned if one channel done
	if (xerr == EALREADY)
		xerr = 0;

	/* Special jail for tail recursion.
	 * Either the data channel closed, or the server has a result for us.
	 * If both are true, notify the caller that the command is done.
	 * Return immediately thereafter -- tail recursion is more palatable
	 * than any other kind.  If this isn't ok, we'll have to
	 * queue callbacks somehow.
	 */
	char statusbuf[256];
	int status;
	status = m_proto.getStatus(statusbuf, sizeof(statusbuf));
	if (status && !STATUS_OK(status) && (m_dfd != -1)) {
		/* If command failed, close any open data connection */
		DPRINT(("ftp_client_pipe_t::notifyPollEvent:notify_app: closing dfd %d\n", m_dfd));
		m_dfd_connecting = false;
		int dfd = m_dfd;
		m_dfd = -1;
		m_poller->del(dfd);
		close(dfd);
	}
	DPRINT(("ftp_client_pipe_t::notifyPollEvent:notify_app: status %d, dfd_conn %d, dfd %d, xerr %d\n", 
		status, m_dfd_connecting, m_dfd, xerr));
	if (xerr || (status && (m_dfd == -1))) {
		/* warning: this usually calls notifyPollEvent, i.e. recursion! */
		call_ftpCmdDone(xerr, status, statusbuf);
		/* no statements after here. */
	}
	return 0;
}

/*----------------------------------------------------------------------
 Callback function.
 When the specified time has elapsed, Sked::runAll calls this method,
 which takes care of any read request posted by a call to notifyPollEvent
 while we were asleep.  Any errors that happen during this call
 are reported via ftpCommandDone().
----------------------------------------------------------------------*/
void ftp_client_pipe_t::skedCallback(clock_t now)
{
	switch (m_state) {
	case IDLE:
		if (m_cfd_connecting) {
			EDPRINT(("ftp_client_pipe_t::skedCallback: connect timeout, calling shutdown()\n"));
			shutdownAndNotify();
			return;
		}
		EDPRINT(("ftp_client_pipe_t::skedCallback: command timeout, calling shutdown(), id %d\n",
			m_datainterface->getID()));
		shutdownAndNotify();
		break;

	case GETTING:
		EDPRINT(("ftp_client_pipe_t::skedCallback: get timeout, calling shutdown()\n"));
		shutdownAndNotify();
		break;

	case SLEEPING:
		DPRINT(("ftp_client_pipe_t::skedCallback: waking up fd %d, events %x\n", m_dfd, m_dfd_events));
		GOTO_STATE("skedCallback", GETTING);
		if (m_dfd_events) {
			short events = m_dfd_events;
			m_dfd_events = 0;
			assert(!eclock_after(m_wakeup, now));
			m_wakeup = now;		/* prevent resleep just in case */
			Poller::PollEvent event;
			event.fd = m_dfd;
			event.revents = events;
			event.client = 0;
			int err = notifyPollEvent(&event);
			if (err) {
				EDPRINT(("ftp_client_pipe_t::skedCallback: errno %d waking up, calling shutdown()\n", err));
				shutdownAndNotify();
			}
		}
		break;

	default:
		assert(false);
	}
}

int ftp_client_pipe_t::kickstart(void)
{
	Poller::PollEvent event;
	event.fd = m_cfd;
	event.revents = POLLOUT|KICKSTART;
	event.client = 0;
	return notifyPollEvent(&event);
}

/*---------------------------------------------------------------------
 Log in to the server.  
 Call getStatus() periodically until it returns nonzero to get whether
 this command succeeded.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::login(const char *username, const char *password)
{
	DPRINT(("ftp_client_pipe_t::login(%s, %s)\n", username, password));
	int err = m_proto.login(username, password);
	if (err) return err;
	return kickstart();
}

/*---------------------------------------------------------------------
 Log out from the server.  This triggers the QUIT command.
 Call getStatus() periodically until it returns nonzero to get whether
 this command succeeded.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::quit()
{
	DPRINT(("ftp_client_pipe_t::quit()\n"));
	int err = m_proto.quit();
	if (err) return err;
	return kickstart();
}

/*---------------------------------------------------------------------
 Change directories.
 If dir is "..", the CDUP command is used instead of CD.
 Call getStatus() periodically until it returns nonzero to get whether
 this command succeeded.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::cd(const char *dir)
{
	DPRINT(("ftp_client_pipe_t::cd(%s)\n", dir));
	int err = m_proto.cd(dir);
	if (err) return err;
	return kickstart();
}

/*---------------------------------------------------------------------
 Set the transfer type.
 Call getStatus() periodically until it returns nonzero to get whether
 this command succeeded, then parse the results out of the status 
 buffer.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::type(const char *ttype)
{
	DPRINT(("ftp_client_pipe_t::type(%s)\n", ttype));
	int err = m_proto.type(ttype);
	if (err) return err;
	return kickstart();
}

/*---------------------------------------------------------------------
 Retrieve file size.
 Call getStatus() periodically until it returns nonzero to get whether
 this command succeeded, then parse the results out of the status 
 buffer.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::size(const char *fname)
{
	DPRINT(("ftp_client_pipe_t::size(%s)\n", fname));
	int err = m_proto.size(fname);
	if (err) return err;
	return kickstart();
}

/*---------------------------------------------------------------------
 List the given directory's contents.   
 If dirname is NULL, the current directory is listed.
 If passive is true, PASV mode is used, else PORT mode is used.
 datainterface->ftpCmdDone() will be called when done.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::ls(const char *dirname, bool passive)
{
	DPRINT(("ftp_client_pipe_t::ls(%s,%d)\n", dirname, passive));

	/* not implemented */
	assert(passive);
	(void) passive;

	int err = m_proto.ls(dirname, 0);
	if (err) return err;
	GOTO_STATE("ls", GETTING);
	return kickstart();
}

/*---------------------------------------------------------------------
 Retrieve the given file's contents.  
 If passive is true, PASV mode is used, else PORT mode is used.
 datainterface->ftpCmdDone() will be called when done.
---------------------------------------------------------------------*/
int ftp_client_pipe_t::get(const char *fname, bool passive)
{
	DPRINT(("ftp_client_pipe_t::get(%s,%d)\n", fname, passive));

	/* not implemented */
	assert(passive);
	(void) passive;

	int err = m_proto.get(fname, NULL);
	if (err) return err;
	GOTO_STATE("get", GETTING);
	m_bytesUnsleptFor = 0;
	m_wakeup = eclock();	/* now */

	/* Abort if no response in five seconds */
	m_sked->addClient(this, m_wakeup + (5 * eclock_hertz()));

	return kickstart();
}

/* other methods' implementations would go here, but haven't been written yet */