<commit_before>/**
 * Appcelerator Titanium - licensed under the Apache Public License 2
 * see LICENSE in the root folder for details on the license.
 * Copyright (c) 2009 Appcelerator, Inc. All Rights Reserved.
 */
 
#include "native_pipe.h"

namespace ti
{
	NativePipe::NativePipe(bool isReader) :
		closed(false),
		isReader(isReader),
		writeThreadAdapter(new Poco::RunnableAdapter<NativePipe>(
			*this, &NativePipe::PollForWrites)),
		readThreadAdapter(new Poco::RunnableAdapter<NativePipe>(
			*this, &NativePipe::PollForReads)),
		readCallback(0),
		logger(Logger::Get("Process.NativePipe"))
	{
	}

	NativePipe::~NativePipe ()
	{
		// Don't need to StopMonitors here, because the destructor
		// should never be called until the monitors are shutdown
		delete readThreadAdapter;
		delete writeThreadAdapter;
	}

	void NativePipe::StopMonitors()
	{
		closed = true;
		try
		{
			if (readThread.isRunning())
					this->readThread.join();
			if (writeThread.isRunning())
					this->writeThread.join();
		}
		catch (Poco::Exception& e)
		{
			logger->Error("Exception while try to join with Pipe thread: %s",
				e.displayText().c_str());
		}
	}

	void NativePipe::Close()
	{
		if (!isReader)
		{
			closed = true;
		}
		Pipe::Close();
	}

	int NativePipe::Write(AutoBlob blob)
	{
		if (isReader)
		{
			// If this is a reader pipe (ie one reading from stdout and stderr
			// via polling), then we want to pass along the data to all attached
			// pipes

			// Someone (probably a process) wants to subscribe to this pipe's
			// reads synchronously. So we need to call the callback on this thread
			// right now.
			if (!readCallback.isNull())
			{
				readCallback->Call(Value::NewObject(blob));
			}

			return Pipe::Write(blob);
		}
		else
		{
			// If this is not a reader pipe (ie one that simply accepts write
			// requests via the Write(...) method, like stdin), then queue the
			// data to be written to the native pipe (blocking operation) by
			// our writer thread.:
			Poco::Mutex::ScopedLock lock(buffersMutex);
			buffers.push(blob);
		}

		return blob->Length();
	}

	void NativePipe::StartMonitor()
	{
		if (isReader)
		{
			readThread.start(*readThreadAdapter);
		}
		else
		{
			writeThread.start(*writeThreadAdapter);
		}
	}

	void NativePipe::PollForReads()
	{
		this->duplicate();

		char buffer[MAX_BUFFER_SIZE];
		int length = MAX_BUFFER_SIZE;
		int bytesRead = this->RawRead(buffer, length);
		while (bytesRead > 0)
		{
			AutoBlob blob = new Blob(buffer, bytesRead);
			this->Write(blob);
			bytesRead = this->RawRead(buffer, length);
		}

		this->release();
	}

	void NativePipe::PollForWrites()
	{
		this->duplicate();

		AutoBlob blob = 0;
		while (!closed || buffers.size() > 0)
		{
			PollForWriteIteration();
		}

		this->CloseNative();
		this->release();
	}

	void NativePipe::PollForWriteIteration()
	{
		AutoBlob blob = 0;
		while (buffers.size() > 0)
		{
			{
				Poco::Mutex::ScopedLock lock(buffersMutex);<commit_msg>Do a little buffering NativePipe::PollForReads to improve Linux performance<commit_after>/**
 * Appcelerator Titanium - licensed under the Apache Public License 2
 * see LICENSE in the root folder for details on the license.
 * Copyright (c) 2009 Appcelerator, Inc. All Rights Reserved.
 */
 
#include "native_pipe.h"
#define READ_LOOP_BUFFER_SIZE 1024

namespace ti
{
	NativePipe::NativePipe(bool isReader) :
		closed(false),
		isReader(isReader),
		writeThreadAdapter(new Poco::RunnableAdapter<NativePipe>(
			*this, &NativePipe::PollForWrites)),
		readThreadAdapter(new Poco::RunnableAdapter<NativePipe>(
			*this, &NativePipe::PollForReads)),
		readCallback(0),
		logger(Logger::Get("Process.NativePipe"))
	{
	}

	NativePipe::~NativePipe ()
	{
		// Don't need to StopMonitors here, because the destructor
		// should never be called until the monitors are shutdown
		delete readThreadAdapter;
		delete writeThreadAdapter;
	}

	void NativePipe::StopMonitors()
	{
		closed = true;
		try
		{
			if (readThread.isRunning())
					this->readThread.join();
			if (writeThread.isRunning())
					this->writeThread.join();
		}
		catch (Poco::Exception& e)
		{
			logger->Error("Exception while try to join with Pipe thread: %s",
				e.displayText().c_str());
		}
	}

	void NativePipe::Close()
	{
		if (!isReader)
		{
			closed = true;
		}
		Pipe::Close();
	}

	int NativePipe::Write(AutoBlob blob)
	{
		if (isReader)
		{
			// If this is a reader pipe (ie one reading from stdout and stderr
			// via polling), then we want to pass along the data to all attached
			// pipes

			// Someone (probably a process) wants to subscribe to this pipe's
			// reads synchronously. So we need to call the callback on this thread
			// right now.
			if (!readCallback.isNull())
			{
				readCallback->Call(Value::NewObject(blob));
			}

			return Pipe::Write(blob);
		}
		else
		{
			// If this is not a reader pipe (ie one that simply accepts write
			// requests via the Write(...) method, like stdin), then queue the
			// data to be written to the native pipe (blocking operation) by
			// our writer thread.:
			Poco::Mutex::ScopedLock lock(buffersMutex);
			buffers.push(blob);
		}

		return blob->Length();
	}

	void NativePipe::StartMonitor()
	{
		if (isReader)
		{
			readThread.start(*readThreadAdapter);
		}
		else
		{
			writeThread.start(*writeThreadAdapter);
		}
	}

	void NativePipe::PollForReads()
	{
		this->duplicate();

		// We want to be somewhat conservative here about when
		// we call this->Write since event handling is inherently
		// slow (it's synchronous and on the main thread). We'll
		// keep a local buffer which we'll periodically glob and
		// push out.
		std::vector<AutoBlob> buffers;
		unsigned int currentBuffersLength = 0;

		char buffer[MAX_BUFFER_SIZE];
		int length = MAX_BUFFER_SIZE;
		int bytesRead = this->RawRead(buffer, length);
		while (bytesRead > 0)
		{
			AutoBlob blob = new Blob(buffer, bytesRead);

			buffers.push_back(blob);
			currentBuffersLength += blob->Length();
			if (currentBuffersLength >= READ_LOOP_BUFFER_SIZE)
			{
				AutoBlob glob = Blob::GlobBlobs(buffers);
				this->Write(glob);
				buffers.clear();
				currentBuffersLength = 0;
			}

			bytesRead = this->RawRead(buffer, length);
		}

		if (!buffers.empty())
		{
			AutoBlob glob = Blob::GlobBlobs(buffers);
			this->Write(glob);
		}

		this->release();
	}

	void NativePipe::PollForWrites()
	{
		this->duplicate();

		AutoBlob blob = 0;
		while (!closed || buffers.size() > 0)
		{
			PollForWriteIteration();
		}

		this->CloseNative();
		this->release();
	}

	void NativePipe::PollForWriteIteration()
	{
		AutoBlob blob = 0;
		while (buffers.size() > 0)
		{
			{
				Poco::Mutex::ScopedLock lock(buffersMutex);
{'fraction_non_alphanumeric': 0.12700074794315633, 'fraction_numerical': 0.004188481675392671, 'mean_word_length': 3.5482993197278914, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 44, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7538256', 'n_tokens_mistral': 2573, 'n_tokens_neox': 2202, 'n_words': 589}