[apple/security.git] / Network / xfercore.cpp

/*
 * Copyright (c) 2000-2001 Apple Computer, Inc. All Rights Reserved.
 * 
 * The contents of this file constitute Original Code as defined in and are
 * subject to the Apple Public Source License Version 1.2 (the 'License').
 * You may not use this file except in compliance with the License. Please obtain
 * a copy of the License at http://www.apple.com/publicsource and read it before
 * using this file.
 * 
 * This Original Code and all software distributed under the License are
 * distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESS
 * OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES, INCLUDING WITHOUT
 * LIMITATION, ANY WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
 * PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT. Please see the License for the
 * specific language governing rights and limitations under the License.
 */


//
// xfercore - core data transfer engine
//
#include "xfercore.h"
#include <Security/debugging.h>


namespace Security {
namespace Network {


//
// Create an engine-level client object.
// @@@ Defer buffer allocation to mating?
// @@@ Defer state initialization to mating?
//
TransferEngine::Client::Client()
    : mMode(invalidInput), mAutoCopyOut(false),
    mSink(NULL), mSource(NULL),
    mAutoFlush(true),
    mReadBuffer(16384), mWriteBuffer(16384)
{
}

TransferEngine::Client::~Client()
{
}


//
// Add and remove clients to/from the engine
//
void TransferEngine::add(Client *client)
{
    client->io = client->fileDesc(); // punch master I/O down to Selector client level
    Selector::add(client->io, *client, input | critical);	// initial registration
}

void TransferEngine::remove(Client *client)
{
#ifndef NDEBUG
    if (!client->mReadBuffer.isEmpty())
        secdebug("xferengine", "xfer %p(%d) HAD %ld BYTES READ LEFT",
            client, client->fileDesc(), client->mReadBuffer.length());
    if (!client->mWriteBuffer.isEmpty())
        secdebug("xferengine", "xfer %p(%d) HAD %ld BYTES WRITE LEFT",
            client, client->fileDesc(), client->mWriteBuffer.length());
#endif //NDEBUG
    if (client->io.fd () != -1) { // did we have a live socket?
        Selector::remove(client->io);
    }

    client->io = FileDesc();	// invalidate
}


//
// Mode switching.
// In addition to the generic switcher (mode), there are variants that set associated
// information, such as sources/sinks.
//
void TransferEngine::Client::mode(InputMode newMode)
{
    secdebug("xferengine", "xfer %p(%d) switching to mode %d", this, fileDesc(), newMode);
    switch (newMode) {
    case rawInput:
    case lineInput:
        mMode = newMode;
        break;
    case connecting:
        enable(output);
        mMode = connecting;
        break;
    default:
        assert(false);	// can't switch to these modes like that
    }
}

void TransferEngine::Client::mode(Sink &sink, size_t byteCount)
{
    mMode = autoReadInput;
    mSink = &sink;
    mResidualReadCount = byteCount;
    secdebug("xferengine", "xfer %p(%d) switching to autoReadInput (%ld bytes)", 
        this, fileDesc(), byteCount);
}

void TransferEngine::Client::mode(Source &source, size_t byteCount)
{
    assert (!mAutoCopyOut);		// no replacements, please
    mAutoCopyOut = true;
    mSource = &source;
    mResidualWriteCount = byteCount;
    secdebug("xferengine", "xfer %p(%d) enabling autoCopyOut mode (%ld bytes)",
        this, fileDesc(), byteCount);
    enable(output);
}


//
// Output methods. This queues output to be sent to the client's connection
// as soon as practical.
//
void TransferEngine::Client::printf(const char *format, ...)
{
    va_list args;
    va_start(args, format);
    vprintf(format, args);
    va_end(args);
}

void TransferEngine::Client::vprintf(const char *format, va_list args)
{
    mWriteBuffer.vprintf(format, args);
#if !defined(NDEBUG)
    char buffer[1024];
    vsnprintf(buffer, sizeof(buffer), format, args);
    secdebug("engineio", "%p(%d) <-- %s", this, fileDesc(), buffer);
#endif //NDEBUG
    startOutput();
}

void TransferEngine::Client::printfe(const char *format, ...)
{
    va_list args;
    va_start(args, format);
    vprintfe(format, args);
    va_end(args);
}

void TransferEngine::Client::vprintfe(const char *format, va_list args)
{
    mWriteBuffer.vprintf(format, args);
    mWriteBuffer.printf("\r\n");
#if !defined(NDEBUG)
    char buffer[1024];
    vsnprintf(buffer, sizeof(buffer), format, args);
    secdebug("engineio", "%p(%d) <-- %s[CRNL]", this, fileDesc(), buffer);
#endif //NDEBUG
    startOutput();
}


//
// Set output auto-flush mode. Think of this as a weak output-hold mode.
// If autoflush is off, we don't try hard to send data out immediately. If it's
// on, we send data as soon as it's generated.
// Calling flushOutput(true) always generates I/O as needed to send output
// data NOW (even if the mode was already on).
// 
void TransferEngine::Client::flushOutput(bool autoFlush)
{
    mAutoFlush = autoFlush;
    secdebug("engineio", "%p(%d) output flush %s", this, fileDesc(), autoFlush? "on" : "off");
    if (mAutoFlush)
        startOutput();
}


//
// StartOutput is called by output generators to get output flowing.
// It may generate output I/O, or hold things in buffers according to
// current settings.
//
void TransferEngine::Client::startOutput()
{
    if (mAutoFlush) {
        if (mAutoCopyOut && !mWriteBuffer.isFull())
            autoCopy();						// try to tack on some autoCopy output
        if (!mWriteBuffer.isEmpty()) {
            mWriteBuffer.write(*this);
            if (mAutoFlush || !mWriteBuffer.isEmpty()) { // possibly more output
                enable(output);				// ask for output-drain notification
            } else {
                disable(output);			// no need for output-possible events
            }
        }
    }
}


//
// Discard any data still in the input buffer.
// This is used to cope with unexpected garbage (protocol violations
// from the server), and shouldn't be used indiscriminately.
//
void TransferEngine::Client::flushInput()
{
    if (!mReadBuffer.isEmpty()) {
        secdebug("engineio", "flushing %ld bytes of input", mReadBuffer.length());
        mReadBuffer.clear();
        mInputFlushed = true;	// inhibit normal buffer ops
    }
}


//
// Given that autoCopyOut mode is active, try to transfer some bytes
// into the write buffer. This is a lazy, fast push, suitable for tacking on
// when you are about to send data for some other reason.
// Returns the number of bytes retrieved from the auto-Source (possibly zero).
//
size_t TransferEngine::Client::autoCopy()
{
    size_t len = mWriteBuffer.available();	//@@@ (true) ?
    if (mResidualWriteCount && mResidualWriteCount < len)
        len = mResidualWriteCount;
    void *addr; mWriteBuffer.locatePut(addr, len);
    mSource->produce(addr, len);
    secdebug("xferengine", "xfer %p(%d) autoCopyOut source delivered %ld bytes",
        this, fileDesc(), len);
    mWriteBuffer.usePut(len);
    return len;
}


//
// This is the notify function called by the IP Selector layer when I/O is possible.
// It runs the state machines for all current clients, calling their transit methods
// in turn.
//
void TransferEngine::Client::notify(int fd, Type type)
{
    try {
        //@@@ Note: We do not currently do anything special about critical events.

        if (type & Selector::output) {
            // if we're in connecting mode
            if (mMode == connecting) {
                Socket s; s = fd;	// Socket(fd) means something different...
                int error = s.error();
                secdebug("xferengine", "xfer %p(%d) connect (errno %d)",
                    this, fd, error);
                transit(connectionDone, NULL, error);
                return;
            }
            
            //@@@ use high/low water marks here
            if (mAutoCopyOut && !mWriteBuffer.isFull()) {
                if (autoCopy() == 0) {
                    switch (mSource->state()) {
                    case Source::stalled:
                        // ah well, maybe later
                        secdebug("xferengine", "xfer %p(%d) autoCopyOut source is stalled", this, fd);
                        break;
                    case Source::endOfData:
                        mAutoCopyOut = false;	// done
                        secdebug("xferengine", "xfer %p(%d) autoCopyOut end of data", this, fd);
                        if (mResidualWriteCount > 0)
                            secdebug("xferengine", "xfer %p(%d) has %ld autoCopy bytes left",
                                this, fd, mResidualWriteCount);
                        transit(autoWriteDone);
                        if (!isActive())
                            return;		// transit removed us; stop now
                        break;
                    default:
                        assert(false);
                    }
                }
            }
            if (mWriteBuffer.isEmpty()) {	// output possible, no output pending
                secdebug("xferengine", "xfer %p(%d) disabling output (empty)", this, fd);
                disable(output);
            } else {					// stuff some more
                size_t length = mWriteBuffer.write(*this);
                secdebug("xferengine", "xfer %p(%d) writing %ld bytes", this, fd, length);
            }
        }
    
        if (type & Selector::input) {
            secdebug("xferengine", "xfer %p(%d) input ready %d bytes",
                this, fd, io.iocget<int>(FIONREAD));
    
            do {
                mInputFlushed = false;	// preset normal
                
                //@@@ break out after partial buffer to give Equal Time to other transfers? good idea?!
                if (!atEnd() && mReadBuffer.read(*this) == 0 && !atEnd()) {
                    mReadBuffer.read(*this, true);
                }
                
                if (mReadBuffer.isEmpty() && atEnd()) {
                    transit(endOfInput);
                    break;
                }
                switch (mMode) {
                case rawInput:
                    rawInputTransit();
                    break;
                case lineInput:
                    if (!lineInputTransit())
                        return;		// no full line; try again later
                    break;
                case autoReadInput:
                    autoReadInputTransit();
                    if (mMode != autoIODone)
                        break;
                    // autoRead completed; fall through to autoIODone handling
                case autoIODone:
                    mMode = invalidInput;		// pre-mark error
                    transit(autoReadDone);		// notify; this must reset mode or exit
                    if (!isActive())			// if we're terminated...
                        return;					// ... then go
                    assert(mMode != invalidInput); // else enforce mode reset
                    break;
                case connecting:
                    {
                        // we should never be here. Selector gave us "read but not write" while connecting. FUBAR
                        Socket s; s = fd;
                        secdebug("xferengine",
                            "fd %d input while connecting (errno=%d, type=%d)",
                            fd, s.error(), type);
                        UnixError::throwMe(ECONNREFUSED);	// likely interpretation
                    }
                default:
                    secdebug("xferengine", "mode error in input sequencer (mode=%d)", mMode);
                    assert(false);
                }
                if (!io)		// client has unhooked; clear buffer and exit loop
                    flushInput();
            } while (!mReadBuffer.isEmpty());
            //@@@ feed back for more output here? But also see comments above...
            //@@@ probably better to take the trip through the Selector
        }
    } catch (const CssmCommonError &err) {
        transitError(err);
    } catch (...) {
        transitError(UnixError::make(EIO));		// best guess (could be anything)
    }
}

void TransferEngine::Client::rawInputTransit()
{
    // just shove it at the user
    char *addr; size_t length = mReadBuffer.length();
    mReadBuffer.locateGet(addr, length);
    secdebug("engineio", "%p(%d) --> %d bytes RAW",
        this, fileDesc(), io.iocget<int>(FIONREAD));
    transit(inputAvailable, addr, length);
    if (!mInputFlushed)
        mReadBuffer.useGet(length);
}

bool TransferEngine::Client::lineInputTransit()
{
    char *line; size_t length = mReadBuffer.length();
    mReadBuffer.locateGet(line, length);

    char *nl;
    for (nl = line; nl < line + length && *nl != '\n'; nl++) ;
    if (nl == line + length)				// no end-of-line, wait for more
        return false;
        
    if (nl > line && nl[-1] == '\r') {		// proper \r\n termination
        nl[-1] = '\0';						// terminate for transit convenience
        secdebug("engineio", "%p(%d) --> %s", this, fileDesc(), line);
        transit(inputAvailable, line, nl - line - 1);
    } else {								// improper, tolerate
        nl[0] = '\0';						// terminate for transit convenience
        secdebug("engineio", "%p(%d) [IMPROPER] --> %s", this, fileDesc(), line);
        transit(inputAvailable, line, nl - line);
    }
    if (!mInputFlushed)
        mReadBuffer.useGet(nl - line + 1);
    return true;
}

void TransferEngine::Client::autoReadInputTransit()
{
    secdebug("xferengine", "xfer %p(%d) %ld pending %d available",
        this, fileDesc(), mReadBuffer.length(), io.iocget<int>(FIONREAD));
    void *data; size_t length = mReadBuffer.length();
    if (mResidualReadCount && mResidualReadCount < length)
        length = mResidualReadCount;
    mReadBuffer.locateGet(data, length);
    secdebug("engineio", "%p(%d) --> %ld bytes autoReadInput", this, fileDesc(), length);
    mSink->consume(data, length);
    if (!mInputFlushed)
        mReadBuffer.useGet(length);
    if (mResidualReadCount && (mResidualReadCount -= length) == 0)
        mMode = autoIODone;
}


//
// The (protected) tickle() method causes a one-time scan
// of the requesting client. This will simulate an input-ready event
// and possibly call the transit method.
// This is designed to be used from validate() or in other unusual
// external situations. Don't call this from within transit().
//
void TransferEngine::Client::tickle()
{
    notify(io, input | critical);
}


//
// The default read/write methods perform direct I/O on the underlying file descriptor.
//
size_t TransferEngine::Client::read(void *data, size_t size)
{ return io.read(data, size); }

size_t TransferEngine::Client::write(const void *data, size_t size)
{ return io.write(data, size); }

bool TransferEngine::Client::atEnd() const
{ return io.atEnd(); }


}	// end namespace Network
}	// end namespace Security
Commit	Line	Data
bac41a7b A	1	/*
	2	* Copyright (c) 2000-2001 Apple Computer, Inc. All Rights Reserved.
	3	*
	4	* The contents of this file constitute Original Code as defined in and are
	5	* subject to the Apple Public Source License Version 1.2 (the 'License').
	6	* You may not use this file except in compliance with the License. Please obtain
	7	* a copy of the License at http://www.apple.com/publicsource and read it before
	8	* using this file.
	9	*
	10	* This Original Code and all software distributed under the License are
	11	* distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESS
	12	* OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES, INCLUDING WITHOUT
	13	* LIMITATION, ANY WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
	14	* PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT. Please see the License for the
	15	* specific language governing rights and limitations under the License.
	16	*/
	17
	18
	19	//
	20	// xfercore - core data transfer engine
	21	//
	22	#include "xfercore.h"
	23	#include <Security/debugging.h>
	24
	25
	26	namespace Security {
	27	namespace Network {
	28
	29
	30	//
	31	// Create an engine-level client object.
	32	// @@@ Defer buffer allocation to mating?
	33	// @@@ Defer state initialization to mating?
	34	//
	35	TransferEngine::Client::Client()
	36	: mMode(invalidInput), mAutoCopyOut(false),
	37	mSink(NULL), mSource(NULL),
	38	mAutoFlush(true),
	39	mReadBuffer(16384), mWriteBuffer(16384)
	40	{
	41	}
	42
	43	TransferEngine::Client::~Client()
	44	{
	45	}
	46
	47
	48	//
	49	// Add and remove clients to/from the engine
	50	//
	51	void TransferEngine::add(Client *client)
	52	{
	53	client->io = client->fileDesc(); // punch master I/O down to Selector client level
	54	Selector::add(client->io, *client, input \| critical); // initial registration
	55	}
	56
	57	void TransferEngine::remove(Client *client)
	58	{
	59	#ifndef NDEBUG
	60	if (!client->mReadBuffer.isEmpty())
df0e469f	61	secdebug("xferengine", "xfer %p(%d) HAD %ld BYTES READ LEFT",
bac41a7b A	62	client, client->fileDesc(), client->mReadBuffer.length());
bac41a7b A	63	if (!client->mWriteBuffer.isEmpty())
df0e469f	64	secdebug("xferengine", "xfer %p(%d) HAD %ld BYTES WRITE LEFT",
bac41a7b A	65	client, client->fileDesc(), client->mWriteBuffer.length());
bac41a7b A	66	#endif //NDEBUG
29654253 A	67	if (client->io.fd () != -1) { // did we have a live socket?
	68	Selector::remove(client->io);
	69	}
	70
bac41a7b A	71	client->io = FileDesc(); // invalidate
	72	}
	73
	74
	75	//
	76	// Mode switching.
	77	// In addition to the generic switcher (mode), there are variants that set associated
	78	// information, such as sources/sinks.
	79	//
	80	void TransferEngine::Client::mode(InputMode newMode)
	81	{
df0e469f	82	secdebug("xferengine", "xfer %p(%d) switching to mode %d", this, fileDesc(), newMode);
bac41a7b A	83	switch (newMode) {
	84	case rawInput:
	85	case lineInput:
	86	mMode = newMode;
	87	break;
	88	case connecting:
	89	enable(output);
	90	mMode = connecting;
	91	break;
	92	default:
	93	assert(false); // can't switch to these modes like that
	94	}
	95	}
	96
	97	void TransferEngine::Client::mode(Sink &sink, size_t byteCount)
	98	{
	99	mMode = autoReadInput;
	100	mSink = &sink;
	101	mResidualReadCount = byteCount;
df0e469f	102	secdebug("xferengine", "xfer %p(%d) switching to autoReadInput (%ld bytes)",
bac41a7b A	103	this, fileDesc(), byteCount);
	104	}
	105
	106	void TransferEngine::Client::mode(Source &source, size_t byteCount)
	107	{
	108	assert (!mAutoCopyOut); // no replacements, please
	109	mAutoCopyOut = true;
	110	mSource = &source;
	111	mResidualWriteCount = byteCount;
df0e469f	112	secdebug("xferengine", "xfer %p(%d) enabling autoCopyOut mode (%ld bytes)",
bac41a7b A	113	this, fileDesc(), byteCount);
	114	enable(output);
	115	}
	116
	117
	118	//
	119	// Output methods. This queues output to be sent to the client's connection
	120	// as soon as practical.
	121	//
	122	void TransferEngine::Client::printf(const char *format, ...)
	123	{
	124	va_list args;
	125	va_start(args, format);
	126	vprintf(format, args);
	127	va_end(args);
	128	}
	129
	130	void TransferEngine::Client::vprintf(const char *format, va_list args)
	131	{
	132	mWriteBuffer.vprintf(format, args);
	133	#if !defined(NDEBUG)
	134	char buffer[1024];
	135	vsnprintf(buffer, sizeof(buffer), format, args);
df0e469f	136	secdebug("engineio", "%p(%d) <-- %s", this, fileDesc(), buffer);
bac41a7b A	137	#endif //NDEBUG
	138	startOutput();
	139	}
	140
	141	void TransferEngine::Client::printfe(const char *format, ...)
	142	{
	143	va_list args;
	144	va_start(args, format);
	145	vprintfe(format, args);
	146	va_end(args);
	147	}
	148
	149	void TransferEngine::Client::vprintfe(const char *format, va_list args)
	150	{
	151	mWriteBuffer.vprintf(format, args);
	152	mWriteBuffer.printf("\r\n");
	153	#if !defined(NDEBUG)
	154	char buffer[1024];
	155	vsnprintf(buffer, sizeof(buffer), format, args);
df0e469f	156	secdebug("engineio", "%p(%d) <-- %s[CRNL]", this, fileDesc(), buffer);
bac41a7b A	157	#endif //NDEBUG
	158	startOutput();
	159	}
	160
	161
	162	//
	163	// Set output auto-flush mode. Think of this as a weak output-hold mode.
	164	// If autoflush is off, we don't try hard to send data out immediately. If it's
	165	// on, we send data as soon as it's generated.
	166	// Calling flushOutput(true) always generates I/O as needed to send output
	167	// data NOW (even if the mode was already on).
	168	//
	169	void TransferEngine::Client::flushOutput(bool autoFlush)
	170	{
	171	mAutoFlush = autoFlush;
df0e469f	172	secdebug("engineio", "%p(%d) output flush %s", this, fileDesc(), autoFlush? "on" : "off");
bac41a7b A	173	if (mAutoFlush)
	174	startOutput();
	175	}
	176
	177
	178	//
	179	// StartOutput is called by output generators to get output flowing.
	180	// It may generate output I/O, or hold things in buffers according to
	181	// current settings.
	182	//
	183	void TransferEngine::Client::startOutput()
	184	{
	185	if (mAutoFlush) {
	186	if (mAutoCopyOut && !mWriteBuffer.isFull())
	187	autoCopy(); // try to tack on some autoCopy output
	188	if (!mWriteBuffer.isEmpty()) {
	189	mWriteBuffer.write(*this);
	190	if (mAutoFlush \|\| !mWriteBuffer.isEmpty()) { // possibly more output
	191	enable(output); // ask for output-drain notification
	192	} else {
	193	disable(output); // no need for output-possible events
	194	}
	195	}
	196	}
	197	}
	198
	199
ded8f8e2 A	200	//
	201	// Discard any data still in the input buffer.
	202	// This is used to cope with unexpected garbage (protocol violations
	203	// from the server), and shouldn't be used indiscriminately.
	204	//
	205	void TransferEngine::Client::flushInput()
	206	{
	207	if (!mReadBuffer.isEmpty()) {
df0e469f	208	secdebug("engineio", "flushing %ld bytes of input", mReadBuffer.length());
ded8f8e2 A	209	mReadBuffer.clear();
	210	mInputFlushed = true; // inhibit normal buffer ops
	211	}
	212	}
	213
	214
bac41a7b A	215	//
	216	// Given that autoCopyOut mode is active, try to transfer some bytes
	217	// into the write buffer. This is a lazy, fast push, suitable for tacking on
	218	// when you are about to send data for some other reason.
	219	// Returns the number of bytes retrieved from the auto-Source (possibly zero).
	220	//
	221	size_t TransferEngine::Client::autoCopy()
	222	{
	223	size_t len = mWriteBuffer.available(); //@@@ (true) ?
	224	if (mResidualWriteCount && mResidualWriteCount < len)
	225	len = mResidualWriteCount;
	226	void *addr; mWriteBuffer.locatePut(addr, len);
	227	mSource->produce(addr, len);
df0e469f	228	secdebug("xferengine", "xfer %p(%d) autoCopyOut source delivered %ld bytes",
bac41a7b A	229	this, fileDesc(), len);
	230	mWriteBuffer.usePut(len);
	231	return len;
	232	}
	233
	234
	235	//
	236	// This is the notify function called by the IP Selector layer when I/O is possible.
	237	// It runs the state machines for all current clients, calling their transit methods
	238	// in turn.
	239	//
	240	void TransferEngine::Client::notify(int fd, Type type)
	241	{
	242	try {
	243	//@@@ Note: We do not currently do anything special about critical events.
	244
	245	if (type & Selector::output) {
	246	// if we're in connecting mode
	247	if (mMode == connecting) {
	248	Socket s; s = fd; // Socket(fd) means something different...
	249	int error = s.error();
df0e469f	250	secdebug("xferengine", "xfer %p(%d) connect (errno %d)",
bac41a7b A	251	this, fd, error);
	252	transit(connectionDone, NULL, error);
	253	return;
	254	}
	255
	256	//@@@ use high/low water marks here
	257	if (mAutoCopyOut && !mWriteBuffer.isFull()) {
	258	if (autoCopy() == 0) {
	259	switch (mSource->state()) {
	260	case Source::stalled:
	261	// ah well, maybe later
df0e469f	262	secdebug("xferengine", "xfer %p(%d) autoCopyOut source is stalled", this, fd);
bac41a7b A	263	break;
	264	case Source::endOfData:
	265	mAutoCopyOut = false; // done
df0e469f	266	secdebug("xferengine", "xfer %p(%d) autoCopyOut end of data", this, fd);
bac41a7b	267	if (mResidualWriteCount > 0)
df0e469f	268	secdebug("xferengine", "xfer %p(%d) has %ld autoCopy bytes left",
bac41a7b A	269	this, fd, mResidualWriteCount);
	270	transit(autoWriteDone);
	271	if (!isActive())
	272	return; // transit removed us; stop now
	273	break;
	274	default:
	275	assert(false);
	276	}
	277	}
	278	}
	279	if (mWriteBuffer.isEmpty()) { // output possible, no output pending
df0e469f	280	secdebug("xferengine", "xfer %p(%d) disabling output (empty)", this, fd);
bac41a7b A	281	disable(output);
	282	} else { // stuff some more
	283	size_t length = mWriteBuffer.write(*this);
df0e469f	284	secdebug("xferengine", "xfer %p(%d) writing %ld bytes", this, fd, length);
bac41a7b A	285	}
	286	}
	287
	288	if (type & Selector::input) {
df0e469f A	289	secdebug("xferengine", "xfer %p(%d) input ready %d bytes",
df0e469f A	290	this, fd, io.iocget<int>(FIONREAD));
bac41a7b A	291
bac41a7b A	292	do {
ded8f8e2 A	293	mInputFlushed = false; // preset normal
ded8f8e2 A	294
bac41a7b	295	//@@@ break out after partial buffer to give Equal Time to other transfers? good idea?!
ded8f8e2	296	if (!atEnd() && mReadBuffer.read(*this) == 0 && !atEnd()) {
bac41a7b A	297	mReadBuffer.read(*this, true);
	298	}
	299
	300	if (mReadBuffer.isEmpty() && atEnd()) {
	301	transit(endOfInput);
	302	break;
	303	}
	304	switch (mMode) {
	305	case rawInput:
	306	rawInputTransit();
	307	break;
	308	case lineInput:
ded8f8e2 A	309	if (!lineInputTransit())
ded8f8e2 A	310	return; // no full line; try again later
bac41a7b A	311	break;
	312	case autoReadInput:
	313	autoReadInputTransit();
	314	if (mMode != autoIODone)
	315	break;
	316	// autoRead completed; fall through to autoIODone handling
	317	case autoIODone:
	318	mMode = invalidInput; // pre-mark error
	319	transit(autoReadDone); // notify; this must reset mode or exit
	320	if (!isActive()) // if we're terminated...
	321	return; // ... then go
	322	assert(mMode != invalidInput); // else enforce mode reset
	323	break;
	324	case connecting:
	325	{
	326	// we should never be here. Selector gave us "read but not write" while connecting. FUBAR
	327	Socket s; s = fd;
df0e469f	328	secdebug("xferengine",
bac41a7b A	329	"fd %d input while connecting (errno=%d, type=%d)",
	330	fd, s.error(), type);
	331	UnixError::throwMe(ECONNREFUSED); // likely interpretation
	332	}
	333	default:
df0e469f	334	secdebug("xferengine", "mode error in input sequencer (mode=%d)", mMode);
bac41a7b A	335	assert(false);
	336	}
	337	if (!io) // client has unhooked; clear buffer and exit loop
ded8f8e2	338	flushInput();
bac41a7b A	339	} while (!mReadBuffer.isEmpty());
	340	//@@@ feed back for more output here? But also see comments above...
	341	//@@@ probably better to take the trip through the Selector
	342	}
29654253	343	} catch (const CssmCommonError &err) {
bac41a7b A	344	transitError(err);
	345	} catch (...) {
	346	transitError(UnixError::make(EIO)); // best guess (could be anything)
	347	}
	348	}
	349
	350	void TransferEngine::Client::rawInputTransit()
	351	{
	352	// just shove it at the user
	353	char *addr; size_t length = mReadBuffer.length();
	354	mReadBuffer.locateGet(addr, length);
df0e469f A	355	secdebug("engineio", "%p(%d) --> %d bytes RAW",
df0e469f A	356	this, fileDesc(), io.iocget<int>(FIONREAD));
bac41a7b	357	transit(inputAvailable, addr, length);
ded8f8e2 A	358	if (!mInputFlushed)
ded8f8e2 A	359	mReadBuffer.useGet(length);
bac41a7b A	360	}
bac41a7b A	361
ded8f8e2	362	bool TransferEngine::Client::lineInputTransit()
bac41a7b A	363	{
	364	char *line; size_t length = mReadBuffer.length();
	365	mReadBuffer.locateGet(line, length);
	366
	367	char *nl;
	368	for (nl = line; nl < line + length && *nl != '\n'; nl++) ;
	369	if (nl == line + length) // no end-of-line, wait for more
ded8f8e2	370	return false;
bac41a7b A	371
	372	if (nl > line && nl[-1] == '\r') { // proper \r\n termination
	373	nl[-1] = '\0'; // terminate for transit convenience
df0e469f	374	secdebug("engineio", "%p(%d) --> %s", this, fileDesc(), line);
bac41a7b A	375	transit(inputAvailable, line, nl - line - 1);
	376	} else { // improper, tolerate
	377	nl[0] = '\0'; // terminate for transit convenience
df0e469f	378	secdebug("engineio", "%p(%d) [IMPROPER] --> %s", this, fileDesc(), line);
bac41a7b A	379	transit(inputAvailable, line, nl - line);
bac41a7b A	380	}
ded8f8e2 A	381	if (!mInputFlushed)
	382	mReadBuffer.useGet(nl - line + 1);
	383	return true;
bac41a7b A	384	}
	385
	386	void TransferEngine::Client::autoReadInputTransit()
	387	{
df0e469f	388	secdebug("xferengine", "xfer %p(%d) %ld pending %d available",
bac41a7b A	389	this, fileDesc(), mReadBuffer.length(), io.iocget<int>(FIONREAD));
	390	void *data; size_t length = mReadBuffer.length();
	391	if (mResidualReadCount && mResidualReadCount < length)
	392	length = mResidualReadCount;
	393	mReadBuffer.locateGet(data, length);
df0e469f	394	secdebug("engineio", "%p(%d) --> %ld bytes autoReadInput", this, fileDesc(), length);
bac41a7b	395	mSink->consume(data, length);
ded8f8e2 A	396	if (!mInputFlushed)
ded8f8e2 A	397	mReadBuffer.useGet(length);
bac41a7b A	398	if (mResidualReadCount && (mResidualReadCount -= length) == 0)
	399	mMode = autoIODone;
	400	}
	401
	402
	403	//
	404	// The (protected) tickle() method causes a one-time scan
	405	// of the requesting client. This will simulate an input-ready event
	406	// and possibly call the transit method.
	407	// This is designed to be used from validate() or in other unusual
	408	// external situations. Don't call this from within transit().
	409	//
	410	void TransferEngine::Client::tickle()
	411	{
	412	notify(io, input \| critical);
	413	}
	414
	415
	416	//
	417	// The default read/write methods perform direct I/O on the underlying file descriptor.
	418	//
	419	size_t TransferEngine::Client::read(void *data, size_t size)
	420	{ return io.read(data, size); }
	421
	422	size_t TransferEngine::Client::write(const void *data, size_t size)
	423	{ return io.write(data, size); }
	424
	425	bool TransferEngine::Client::atEnd() const
	426	{ return io.atEnd(); }
	427
	428
	429	} // end namespace Network
	430	} // end namespace Security