dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Add Process.runSync for running processes synchronously

Browse Source 
BUG=http://dartbug.com/1707

R=whesse@google.com

Review URL: https://codereview.chromium.org//21816002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@26052 260f80e4-7a28-3924-810f-c04153c831b5
This commit is contained in:
sgjesse@google.com
2013-08-13 14:12:31 +00:00
parent 48cf6c5ebd
commit 959456533f
14 changed files with 894 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/bin/eventhandler_win.cc
+53 -49
View File
@@ -26,51 +26,54 @@ namespace dart {
namespace bin {
static const int kBufferSize = 64 * 1024;
static const int kStdioBufferSize = 16 * 1024;
static const int kStdOverlappedBufferSize = 16 * 1024;
static const int kInfinityTimeout = -1;
static const int kTimeoutId = -1;
static const int kShutdownId = -2;
IOBuffer* IOBuffer::AllocateBuffer(int buffer_size, Operation operation) {
IOBuffer* buffer = new(buffer_size) IOBuffer(buffer_size, operation);
OverlappedBuffer* OverlappedBuffer::AllocateBuffer(int buffer_size,
Operation operation) {
OverlappedBuffer* buffer =
new(buffer_size) OverlappedBuffer(buffer_size, operation);
return buffer;
}
IOBuffer* IOBuffer::AllocateAcceptBuffer(int buffer_size) {
IOBuffer* buffer = AllocateBuffer(buffer_size, kAccept);
OverlappedBuffer* OverlappedBuffer::AllocateAcceptBuffer(int buffer_size) {
OverlappedBuffer* buffer = AllocateBuffer(buffer_size, kAccept);
return buffer;
}
IOBuffer* IOBuffer::AllocateReadBuffer(int buffer_size) {
OverlappedBuffer* OverlappedBuffer::AllocateReadBuffer(int buffer_size) {
return AllocateBuffer(buffer_size, kRead);
}
IOBuffer* IOBuffer::AllocateWriteBuffer(int buffer_size) {
OverlappedBuffer* OverlappedBuffer::AllocateWriteBuffer(int buffer_size) {
return AllocateBuffer(buffer_size, kWrite);
}
IOBuffer* IOBuffer::AllocateDisconnectBuffer() {
OverlappedBuffer* OverlappedBuffer::AllocateDisconnectBuffer() {
return AllocateBuffer(0, kDisconnect);
}
void IOBuffer::DisposeBuffer(IOBuffer* buffer) {
void OverlappedBuffer::DisposeBuffer(OverlappedBuffer* buffer) {
delete buffer;
}
IOBuffer* IOBuffer::GetFromOverlapped(OVERLAPPED* overlapped) {
IOBuffer* buffer = CONTAINING_RECORD(overlapped, IOBuffer, overlapped_);
OverlappedBuffer* OverlappedBuffer::GetFromOverlapped(OVERLAPPED* overlapped) {
OverlappedBuffer* buffer =
CONTAINING_RECORD(overlapped, OverlappedBuffer, overlapped_);
return buffer;
}
int IOBuffer::Read(void* buffer, int num_bytes) {
int OverlappedBuffer::Read(void* buffer, int num_bytes) {
if (num_bytes > GetRemainingLength()) {
num_bytes = GetRemainingLength();
}
@@ -80,7 +83,7 @@ int IOBuffer::Read(void* buffer, int num_bytes) {
}
int IOBuffer::Write(const void* buffer, int num_bytes) {
int OverlappedBuffer::Write(const void* buffer, int num_bytes) {
ASSERT(num_bytes == buflen_);
memcpy(GetBufferStart(), buffer, num_bytes);
data_length_ = num_bytes;
@@ -88,7 +91,7 @@ int IOBuffer::Write(const void* buffer, int num_bytes) {
}
int IOBuffer::GetRemainingLength() {
int OverlappedBuffer::GetRemainingLength() {
ASSERT(operation_ == kRead);
return data_length_ - index_;
}
@@ -183,7 +186,7 @@ bool Handle::HasPendingWrite() {
}
void Handle::ReadComplete(IOBuffer* buffer) {
void Handle::ReadComplete(OverlappedBuffer* buffer) {
ScopedLock lock(this);
// Currently only one outstanding read at the time.
ASSERT(pending_read_ == buffer);
@@ -191,17 +194,17 @@ void Handle::ReadComplete(IOBuffer* buffer) {
if (!IsClosing() && !buffer->IsEmpty()) {
data_ready_ = pending_read_;
} else {
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
}
pending_read_ = NULL;
}
void Handle::WriteComplete(IOBuffer* buffer) {
void Handle::WriteComplete(OverlappedBuffer* buffer) {
ScopedLock lock(this);
// Currently only one outstanding write at the time.
ASSERT(pending_write_ == buffer);
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
pending_write_ = NULL;
}
@@ -215,11 +218,11 @@ static unsigned int __stdcall ReadFileThread(void* args) {
void Handle::ReadSyncCompleteAsync() {
ASSERT(pending_read_ != NULL);
ASSERT(pending_read_->GetBufferSize() >= kStdioBufferSize);
ASSERT(pending_read_->GetBufferSize() >= kStdOverlappedBufferSize);
DWORD buffer_size = pending_read_->GetBufferSize();
if (GetFileType(handle_) == FILE_TYPE_CHAR) {
buffer_size = kStdioBufferSize;
buffer_size = kStdOverlappedBufferSize;
}
DWORD bytes_read = 0;
BOOL ok = ReadFile(handle_,
@@ -248,7 +251,7 @@ bool Handle::IssueRead() {
ScopedLock lock(this);
ASSERT(type_ != kListenSocket);
ASSERT(pending_read_ == NULL);
IOBuffer* buffer = IOBuffer::AllocateReadBuffer(kBufferSize);
OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
if (SupportsOverlappedIO()) {
ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
@@ -262,7 +265,7 @@ bool Handle::IssueRead() {
pending_read_ = buffer;
return true;
}
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
HandleIssueError();
return false;
} else {
@@ -284,9 +287,9 @@ bool Handle::IssueWrite() {
ASSERT(type_ != kListenSocket);
ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
ASSERT(pending_write_ != NULL);
ASSERT(pending_write_->operation() == IOBuffer::kWrite);
ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite);
IOBuffer* buffer = pending_write_;
OverlappedBuffer* buffer = pending_write_;
BOOL ok = WriteFile(handle_,
buffer->GetBufferStart(),
buffer->GetBufferSize(),
@@ -297,7 +300,7 @@ bool Handle::IssueWrite() {
pending_write_ = buffer;
return true;
}
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
HandleIssueError();
return false;
}
@@ -383,8 +386,8 @@ bool ListenSocket::IssueAccept() {
static const int kAcceptExAddressAdditionalBytes = 16;
static const int kAcceptExAddressStorageSize =
sizeof(SOCKADDR_STORAGE) + kAcceptExAddressAdditionalBytes;
IOBuffer* buffer =
IOBuffer::AllocateAcceptBuffer(2 * kAcceptExAddressStorageSize);
OverlappedBuffer* buffer =
OverlappedBuffer::AllocateAcceptBuffer(2 * kAcceptExAddressStorageSize);
DWORD received;
BOOL ok;
ok = AcceptEx_(socket(),
@@ -399,7 +402,7 @@ bool ListenSocket::IssueAccept() {
if (WSAGetLastError() != WSA_IO_PENDING) {
Log::PrintErr("AcceptEx failed: %d\n", WSAGetLastError());
closesocket(buffer->client());
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
return false;
}
}
@@ -410,7 +413,8 @@ bool ListenSocket::IssueAccept() {
}
void ListenSocket::AcceptComplete(IOBuffer* buffer, HANDLE completion_port) {
void ListenSocket::AcceptComplete(OverlappedBuffer* buffer,
HANDLE completion_port) {
ScopedLock lock(this);
if (!IsClosing()) {
// Update the accepted socket to support the full range of API calls.
@@ -441,7 +445,7 @@ void ListenSocket::AcceptComplete(IOBuffer* buffer, HANDLE completion_port) {
}
pending_accept_count_--;
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
}
@@ -508,7 +512,7 @@ int Handle::Read(void* buffer, int num_bytes) {
if (data_ready_ == NULL) return 0;
num_bytes = data_ready_->Read(buffer, num_bytes);
if (data_ready_->IsEmpty()) {
IOBuffer::DisposeBuffer(data_ready_);
OverlappedBuffer::DisposeBuffer(data_ready_);
data_ready_ = NULL;
}
return num_bytes;
@@ -521,7 +525,7 @@ int Handle::Write(const void* buffer, int num_bytes) {
if (pending_write_ != NULL) return 0;
if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
if (num_bytes > kBufferSize) num_bytes = kBufferSize;
pending_write_ = IOBuffer::AllocateWriteBuffer(num_bytes);
pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
pending_write_->Write(buffer, num_bytes);
if (!IssueWrite()) return -1;
return num_bytes;
@@ -587,7 +591,7 @@ bool ClientSocket::IssueRead() {
ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
ASSERT(pending_read_ == NULL);
IOBuffer* buffer = IOBuffer::AllocateReadBuffer(1024);
OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(1024);
DWORD flags;
flags = 0;
@@ -602,7 +606,7 @@ bool ClientSocket::IssueRead() {
pending_read_ = buffer;
return true;
}
IOBuffer::DisposeBuffer(buffer);
OverlappedBuffer::DisposeBuffer(buffer);
pending_read_ = NULL;
HandleIssueError();
return false;
@@ -613,7 +617,7 @@ bool ClientSocket::IssueWrite() {
ScopedLock lock(this);
ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
ASSERT(pending_write_ != NULL);
ASSERT(pending_write_->operation() == IOBuffer::kWrite);
ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite);
int rc = WSASend(socket(),
pending_write_->GetWASBUF(),
@@ -625,7 +629,7 @@ bool ClientSocket::IssueWrite() {
if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
return true;
}
IOBuffer::DisposeBuffer(pending_write_);
OverlappedBuffer::DisposeBuffer(pending_write_);
pending_write_ = NULL;
HandleIssueError();
return false;
@@ -634,7 +638,7 @@ bool ClientSocket::IssueWrite() {
void ClientSocket::IssueDisconnect() {
Dart_Port p = port();
IOBuffer* buffer = IOBuffer::AllocateDisconnectBuffer();
OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer();
BOOL ok = DisconnectEx_(
socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0);
if (!ok && WSAGetLastError() != WSA_IO_PENDING) {
@@ -644,11 +648,11 @@ void ClientSocket::IssueDisconnect() {
}
void ClientSocket::DisconnectComplete(IOBuffer* buffer) {
IOBuffer::DisposeBuffer(buffer);
void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
OverlappedBuffer::DisposeBuffer(buffer);
closesocket(socket());
if (data_ready_ != NULL) {
IOBuffer::DisposeBuffer(data_ready_);
OverlappedBuffer::DisposeBuffer(data_ready_);
}
// When disconnect is complete get rid of the object.
delete this;
@@ -776,7 +780,7 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket,
IOBuffer* buffer) {
OverlappedBuffer* buffer) {
listen_socket->AcceptComplete(buffer, completion_port_);
if (!listen_socket->IsClosing()) {
@@ -810,7 +814,7 @@ void EventHandlerImplementation::HandleError(Handle* handle) {
void EventHandlerImplementation::HandleRead(Handle* handle,
int bytes,
IOBuffer* buffer) {
OverlappedBuffer* buffer) {
buffer->set_data_length(bytes);
handle->ReadComplete(buffer);
if (bytes > 0) {
@@ -835,7 +839,7 @@ void EventHandlerImplementation::HandleRead(Handle* handle,
void EventHandlerImplementation::HandleWrite(Handle* handle,
int bytes,
IOBuffer* buffer) {
OverlappedBuffer* buffer) {
handle->WriteComplete(buffer);
if (bytes > 0) {
@@ -858,7 +862,7 @@ void EventHandlerImplementation::HandleWrite(Handle* handle,
void EventHandlerImplementation::HandleDisconnect(
ClientSocket* client_socket,
int bytes,
IOBuffer* buffer) {
OverlappedBuffer* buffer) {
client_socket->DisconnectComplete(buffer);
}
@@ -872,24 +876,24 @@ void EventHandlerImplementation::HandleTimeout() {
void EventHandlerImplementation::HandleIOCompletion(DWORD bytes,
ULONG_PTR key,
OVERLAPPED* overlapped) {
IOBuffer* buffer = IOBuffer::GetFromOverlapped(overlapped);
OverlappedBuffer* buffer = OverlappedBuffer::GetFromOverlapped(overlapped);
switch (buffer->operation()) {
case IOBuffer::kAccept: {
case OverlappedBuffer::kAccept: {
ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(key);
HandleAccept(listen_socket, buffer);
break;
}
case IOBuffer::kRead: {
case OverlappedBuffer::kRead: {
Handle* handle = reinterpret_cast<Handle*>(key);
HandleRead(handle, bytes, buffer);
break;
}
case IOBuffer::kWrite: {
case OverlappedBuffer::kWrite: {
Handle* handle = reinterpret_cast<Handle*>(key);
HandleWrite(handle, bytes, buffer);
break;
}
case IOBuffer::kDisconnect: {
case OverlappedBuffer::kDisconnect: {
ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
HandleDisconnect(client_socket, bytes, buffer);
break;
runtime/bin/eventhandler_win.h
+22 -21
View File
@@ -34,21 +34,21 @@ struct InterruptMessage {
};
// An IOBuffer encapsulates the OVERLAPPED structure and the
// An OverlappedBuffer encapsulates the OVERLAPPED structure and the
// associated data buffer. For accept it also contains the pre-created
// socket for the client.
class IOBuffer {
class OverlappedBuffer {
public:
enum Operation { kAccept, kRead, kWrite, kDisconnect };
static IOBuffer* AllocateAcceptBuffer(int buffer_size);
static IOBuffer* AllocateReadBuffer(int buffer_size);
static IOBuffer* AllocateWriteBuffer(int buffer_size);
static IOBuffer* AllocateDisconnectBuffer();
static void DisposeBuffer(IOBuffer* buffer);
static OverlappedBuffer* AllocateAcceptBuffer(int buffer_size);
static OverlappedBuffer* AllocateReadBuffer(int buffer_size);
static OverlappedBuffer* AllocateWriteBuffer(int buffer_size);
static OverlappedBuffer* AllocateDisconnectBuffer();
static void DisposeBuffer(OverlappedBuffer* buffer);
// Find the IO buffer from the OVERLAPPED address.
static IOBuffer* GetFromOverlapped(OVERLAPPED* overlapped);
static OverlappedBuffer* GetFromOverlapped(OVERLAPPED* overlapped);
// Read data from a buffer which has been received. It will read up
// to num_bytes bytes of data returning the actual number of bytes
@@ -88,7 +88,7 @@ class IOBuffer {
void set_data_length(int data_length) { data_length_ = data_length; }
private:
IOBuffer(int buffer_size, Operation operation)
OverlappedBuffer(int buffer_size, Operation operation)
: operation_(operation), buflen_(buffer_size) {
memset(GetBufferStart(), 0, GetBufferSize());
index_ = 0;
@@ -106,7 +106,8 @@ class IOBuffer {
free(buffer);
}
static IOBuffer* AllocateBuffer(int buffer_size, Operation operation);
static OverlappedBuffer* AllocateBuffer(int buffer_size,
Operation operation);
OVERLAPPED overlapped_; // OVERLAPPED structure for overlapped IO.
SOCKET client_; // Used for AcceptEx client socket.
@@ -157,8 +158,8 @@ class Handle {
virtual bool IssueWrite();
bool HasPendingRead();
bool HasPendingWrite();
void ReadComplete(IOBuffer* buffer);
void WriteComplete(IOBuffer* buffer);
void ReadComplete(OverlappedBuffer* buffer);
void WriteComplete(OverlappedBuffer* buffer);
bool IsClosing() { return (flags_ & (1 << kClosing)) != 0; }
bool IsClosedRead() { return (flags_ & (1 << kCloseRead)) != 0; }
@@ -230,9 +231,9 @@ class Handle {
HANDLE completion_port_;
EventHandlerImplementation* event_handler_;
IOBuffer* data_ready_; // IO buffer for data ready to be read.
IOBuffer* pending_read_; // IO buffer for pending read.
IOBuffer* pending_write_; // IO buffer for pending write
OverlappedBuffer* data_ready_; // Buffer for data ready to be read.
OverlappedBuffer* pending_read_; // Buffer for pending read.
OverlappedBuffer* pending_write_; // Buffer for pending write
DWORD last_error_;
@@ -291,7 +292,7 @@ class ListenSocket : public SocketHandle {
// Internal interface used by the event handler.
bool HasPendingAccept() { return pending_accept_count_ > 0; }
bool IssueAccept();
void AcceptComplete(IOBuffer* buffer, HANDLE completion_port);
void AcceptComplete(OverlappedBuffer* buffer, HANDLE completion_port);
virtual void EnsureInitialized(
EventHandlerImplementation* event_handler);
@@ -342,7 +343,7 @@ class ClientSocket : public SocketHandle {
virtual bool IssueRead();
virtual bool IssueWrite();
void IssueDisconnect();
void DisconnectComplete(IOBuffer* buffer);
void DisconnectComplete(OverlappedBuffer* buffer);
virtual void EnsureInitialized(
EventHandlerImplementation* event_handler);
@@ -375,14 +376,14 @@ class EventHandlerImplementation {
int64_t GetTimeout();
void HandleInterrupt(InterruptMessage* msg);
void HandleTimeout();
void HandleAccept(ListenSocket* listen_socket, IOBuffer* buffer);
void HandleAccept(ListenSocket* listen_socket, OverlappedBuffer* buffer);
void HandleClosed(Handle* handle);
void HandleError(Handle* handle);
void HandleRead(Handle* handle, int bytes, IOBuffer* buffer);
void HandleWrite(Handle* handle, int bytes, IOBuffer* buffer);
void HandleRead(Handle* handle, int bytes, OverlappedBuffer* buffer);
void HandleWrite(Handle* handle, int bytes, OverlappedBuffer* buffer);
void HandleDisconnect(ClientSocket* client_socket,
int bytes,
IOBuffer* buffer);
OverlappedBuffer* buffer);
void HandleIOCompletion(DWORD bytes, ULONG_PTR key, OVERLAPPED* overlapped);
HANDLE completion_port() { return completion_port_; }
runtime/bin/io_buffer.cc
+1
View File
@@ -24,6 +24,7 @@ Dart_Handle IOBuffer::Allocate(intptr_t size, uint8_t **buffer) {
return result;
}
uint8_t* IOBuffer::Allocate(intptr_t size) {
return new uint8_t[size];
}
runtime/bin/io_natives.cc
+1
View File
@@ -37,6 +37,7 @@ namespace bin {
V(Platform_Environment, 0) \
V(Platform_GetVersion, 0) \
V(Process_Start, 10) \
V(Process_Wait, 5) \
V(Process_Kill, 3) \
V(Process_SetExitCode, 1) \
V(Process_Exit, 1) \
runtime/bin/process.cc
+42
View File
@@ -159,6 +159,48 @@ void FUNCTION_NAME(Process_Start)(Dart_NativeArguments args) {
}
void FUNCTION_NAME(Process_Wait)(Dart_NativeArguments args) {
Dart_Handle process = Dart_GetNativeArgument(args, 0);
Dart_Handle stdin_handle = Dart_GetNativeArgument(args, 1);
Dart_Handle stdout_handle = Dart_GetNativeArgument(args, 2);
Dart_Handle stderr_handle = Dart_GetNativeArgument(args, 3);
Dart_Handle exit_handle = Dart_GetNativeArgument(args, 4);
intptr_t process_stdin;
intptr_t process_stdout;
intptr_t process_stderr;
intptr_t exit_event;
Socket::GetSocketIdNativeField(stdin_handle, &process_stdin);
Socket::GetSocketIdNativeField(stdout_handle, &process_stdout);
Socket::GetSocketIdNativeField(stderr_handle, &process_stderr);
Socket::GetSocketIdNativeField(exit_handle, &exit_event);
ProcessResult result;
intptr_t pid;
Process::GetProcessIdNativeField(process, &pid);
if (Process::Wait(pid,
process_stdin,
process_stdout,
process_stderr,
exit_event,
&result)) {
Dart_Handle out = result.stdout_data();
if (Dart_IsError(out)) Dart_PropagateError(out);
Dart_Handle err = result.stderr_data();
if (Dart_IsError(err)) Dart_PropagateError(err);
Dart_Handle list = Dart_NewList(4);
Dart_ListSetAt(list, 0, Dart_NewInteger(pid));
Dart_ListSetAt(list, 1, Dart_NewInteger(result.exit_code()));
Dart_ListSetAt(list, 2, out);
Dart_ListSetAt(list, 3, err);
Dart_SetReturnValue(args, list);
} else {
Dart_Handle error = DartUtils::NewDartOSError();
Process::Kill(pid, 9);
if (Dart_IsError(error)) Dart_PropagateError(error);
Dart_ThrowException(error);
}
}
void FUNCTION_NAME(Process_Kill)(Dart_NativeArguments args) {
Dart_Handle process = Dart_GetNativeArgument(args, 1);
intptr_t pid = -1;
runtime/bin/process.h
+137
View File
@@ -6,13 +6,41 @@
#define BIN_PROCESS_H_
#include "bin/builtin.h"
#include "bin/io_buffer.h"
#include "bin/thread.h"
#include "platform/globals.h"
#include "platform/utils.h"
namespace dart {
namespace bin {
class ProcessResult {
public:
ProcessResult() : exit_code_(0) {}
void set_stdout_data(Dart_Handle stdout_data) {
stdout_data_ = stdout_data;
}
void set_stderr_data(Dart_Handle stderr_data) {
stderr_data_ = stderr_data;
}
void set_exit_code(intptr_t exit_code) { exit_code_ = exit_code; }
Dart_Handle stdout_data() { return stdout_data_; }
Dart_Handle stderr_data() { return stderr_data_; }
intptr_t exit_code() { return exit_code_; }
private:
Dart_Handle stdout_data_;
Dart_Handle stderr_data_;
intptr_t exit_code_;
DISALLOW_ALLOCATION();
};
class Process {
public:
// Start a new process providing access to stdin, stdout, stderr and
@@ -30,6 +58,13 @@ class Process {
intptr_t* exit_handler,
char** os_error_message);
static bool Wait(intptr_t id,
intptr_t in,
intptr_t out,
intptr_t err,
intptr_t exit_handler,
ProcessResult* result);
// Kill a process with a given pid.
static bool Kill(intptr_t id, int signal);
@@ -62,6 +97,108 @@ class Process {
DISALLOW_IMPLICIT_CONSTRUCTORS(Process);
};
// Utility class for collecting the output when running a process
// synchronously by using Process::Wait. This class is sub-classed in
// the platform specific files to implement reading into the buffers
// allocated.
class BufferListBase {
protected:
static const intptr_t kBufferSize = 16 * 1024;
class BufferListNode {
public:
explicit BufferListNode(intptr_t size) {
data_ = new uint8_t[size];
if (data_ == NULL) FATAL("Allocation failed");
next_ = NULL;
}
~BufferListNode() {
delete[] data_;
}
uint8_t* data_;
BufferListNode* next_;
private:
DISALLOW_IMPLICIT_CONSTRUCTORS(BufferListNode);
};
public:
BufferListBase() : head_(NULL), tail_(NULL), data_size_(0), free_size_(0) {}
~BufferListBase() {
ASSERT(head_ == NULL);
ASSERT(tail_ == NULL);
}
// Returns the collected data as a Uint8List. If an error occours an
// error handle is returned.
Dart_Handle GetData() {
uint8_t* buffer;
intptr_t buffer_position = 0;
Dart_Handle result = IOBuffer::Allocate(data_size_, &buffer);
if (Dart_IsError(result)) {
Free();
return result;
}
for (BufferListNode* current = head_;
current != NULL;
current = current->next_) {
intptr_t to_copy = dart::Utils::Minimum(data_size_, kBufferSize);
memmove(buffer + buffer_position, current->data_, to_copy);
buffer_position += to_copy;
data_size_ -= to_copy;
}
ASSERT(data_size_ == 0);
Free();
return result;
}
protected:
void Allocate() {
ASSERT(free_size_ == 0);
BufferListNode* node = new BufferListNode(kBufferSize);
if (head_ == NULL) {
head_ = node;
tail_ = node;
} else {
ASSERT(tail_->next_ == NULL);
tail_->next_ = node;
tail_ = node;
}
free_size_ = kBufferSize;
}
void Free() {
BufferListNode* current = head_;
while (current != NULL) {
BufferListNode* tmp = current;
current = current->next_;
delete tmp;
}
head_ = NULL;
tail_ = NULL;
data_size_ = 0;
free_size_ = 0;
}
// Returns the address of the first byte in the free space.
uint8_t* FreeSpaceAddress() {
return tail_->data_ + (kBufferSize - free_size_);
}
// Linked list for data collected.
BufferListNode* head_;
BufferListNode* tail_;
// Number of bytes of data collected in the linked list.
intptr_t data_size_;
// Number of free bytes in the last node in the list.
intptr_t free_size_;
};
} // namespace bin
} // namespace dart
runtime/bin/process_linux.cc
+121
View File
@@ -21,6 +21,7 @@
#include "bin/log.h"
#include "bin/thread.h"
extern char **environ;
@@ -568,6 +569,126 @@ int Process::Start(const char* path,
}
class BufferList: public BufferListBase {
public:
bool Read(int fd, intptr_t available) {
// Read all available bytes.
while (available > 0) {
if (free_size_ == 0) Allocate();
ASSERT(free_size_ > 0);
ASSERT(free_size_ <= kBufferSize);
intptr_t block_size = dart::Utils::Minimum(free_size_, available);
intptr_t bytes = TEMP_FAILURE_RETRY(read(
fd,
reinterpret_cast<void*>(FreeSpaceAddress()),
block_size));
if (bytes < 0) return false;
data_size_ += bytes;
free_size_ -= bytes;
available -= bytes;
}
return true;
}
};
static bool CloseProcessBuffers(struct pollfd fds[3]) {
int e = errno;
VOID_TEMP_FAILURE_RETRY(close(fds[0].fd));
VOID_TEMP_FAILURE_RETRY(close(fds[1].fd));
VOID_TEMP_FAILURE_RETRY(close(fds[2].fd));
errno = e;
return false;
}
bool Process::Wait(intptr_t pid,
intptr_t in,
intptr_t out,
intptr_t err,
intptr_t exit_event,
ProcessResult* result) {
// Close input to the process right away.
VOID_TEMP_FAILURE_RETRY(close(in));
// There is no return from this function using Dart_PropagateError
// as memory used by the buffer lists is freed through their
// destructors.
BufferList out_data;
BufferList err_data;
union {
uint8_t bytes[8];
int32_t ints[2];
} exit_code_data;
struct pollfd fds[3];
fds[0].fd = out;
fds[1].fd = err;
fds[2].fd = exit_event;
for (int i = 0; i < 3; i++) {
fds[i].events = POLLIN;
}
int alive = 3;
while (alive > 0) {
// Blocking call waiting for events from the child process.
if (TEMP_FAILURE_RETRY(poll(fds, alive, -1)) <= 0) {
return CloseProcessBuffers(fds);
}
// Process incoming data.
for (int i = 0; i < alive; i++) {
if (fds[i].revents & POLLIN) {
intptr_t avail = FDUtils::AvailableBytes(fds[i].fd);
if (fds[i].fd == out) {
if (!out_data.Read(out, avail)) {
return CloseProcessBuffers(fds);
}
} else if (fds[i].fd == err) {
if (!err_data.Read(err, avail)) {
return CloseProcessBuffers(fds);
}
} else if (fds[i].fd == exit_event) {
if (avail == 8) {
intptr_t b = TEMP_FAILURE_RETRY(read(exit_event,
exit_code_data.bytes, 8));
if (b != 8) {
return CloseProcessBuffers(fds);
}
}
} else {
UNREACHABLE();
}
}
}
// Process closed.
for (int i = 0; i < alive; i++) {
if (fds[i].revents & POLLHUP) {
VOID_TEMP_FAILURE_RETRY(close(fds[i].fd));
alive--;
if (i < alive) {
fds[i] = fds[alive];
}
}
}
}
// All handles closed and all data read.
result->set_stdout_data(out_data.GetData());
result->set_stderr_data(err_data.GetData());
// Calculate the exit code.
intptr_t exit_code = exit_code_data.ints[0];
intptr_t negative = exit_code_data.ints[1];
if (negative) exit_code = -exit_code;
result->set_exit_code(exit_code);
return true;
}
bool Process::Kill(intptr_t id, int signal) {
return (TEMP_FAILURE_RETRY(kill(id, signal)) != -1);
}
runtime/bin/process_macos.cc
+124
View File
@@ -567,6 +567,130 @@ int Process::Start(const char* path,
}
class BufferList: public BufferListBase {
public:
bool Read(int fd, intptr_t available) {
// Read all available bytes.
while (available > 0) {
if (free_size_ == 0) Allocate();
ASSERT(free_size_ > 0);
ASSERT(free_size_ <= kBufferSize);
intptr_t block_size = dart::Utils::Minimum(free_size_, available);
intptr_t bytes = TEMP_FAILURE_RETRY(read(
fd,
reinterpret_cast<void*>(FreeSpaceAddress()),
block_size));
if (bytes < 0) return false;
data_size_ += bytes;
free_size_ -= bytes;
available -= bytes;
}
return true;
}
};
static bool CloseProcessBuffers(struct pollfd fds[3]) {
int e = errno;
VOID_TEMP_FAILURE_RETRY(close(fds[0].fd));
VOID_TEMP_FAILURE_RETRY(close(fds[1].fd));
VOID_TEMP_FAILURE_RETRY(close(fds[2].fd));
errno = e;
return false;
}
bool Process::Wait(intptr_t pid,
intptr_t in,
intptr_t out,
intptr_t err,
intptr_t exit_event,
ProcessResult* result) {
// Close input to the process right away.
VOID_TEMP_FAILURE_RETRY(close(in));
// There is no return from this function using Dart_PropagateError
// as memory used by the buffer lists is freed through their
// destructors.
BufferList out_data;
BufferList err_data;
union {
uint8_t bytes[8];
int32_t ints[2];
} exit_code_data;
struct pollfd fds[3];
fds[0].fd = out;
fds[1].fd = err;
fds[2].fd = exit_event;
for (int i = 0; i < 3; i++) {
fds[i].events = POLLIN;
}
int alive = 3;
while (alive > 0) {
// Blocking call waiting for events from the child process.
if (TEMP_FAILURE_RETRY(poll(fds, alive, -1)) <= 0) {
return CloseProcessBuffers(fds);
}
// Process incoming data.
for (int i = 0; i < alive; i++) {
if (fds[i].revents & POLLIN) {
intptr_t avail = FDUtils::AvailableBytes(fds[i].fd);
// On Mac OS POLLIN can be set with zero available
// bytes. POLLHUP is most likely also set in this case.
if (avail > 0) {
if (fds[i].fd == out) {
if (!out_data.Read(out, avail)) {
return CloseProcessBuffers(fds);
}
} else if (fds[i].fd == err) {
if (!err_data.Read(err, avail)) {
return CloseProcessBuffers(fds);
}
} else if (fds[i].fd == exit_event) {
if (avail == 8) {
intptr_t b = TEMP_FAILURE_RETRY(read(fds[i].fd,
exit_code_data.bytes, 8));
if (b != 8) {
return CloseProcessBuffers(fds);
}
}
} else {
UNREACHABLE();
}
}
}
}
// Process closed.
for (int i = 0; i < alive; i++) {
if (fds[i].revents & POLLHUP) {
VOID_TEMP_FAILURE_RETRY(close(fds[i].fd));
alive--;
if (i < alive) {
fds[i] = fds[alive];
}
}
}
}
// All handles closed and all data read.
result->set_stdout_data(out_data.GetData());
result->set_stderr_data(err_data.GetData());
// Calculate the exit code.
intptr_t exit_code = exit_code_data.ints[0];
intptr_t negative = exit_code_data.ints[1];
if (negative) exit_code = -exit_code;
result->set_exit_code(exit_code);
return true;
}
bool Process::Kill(intptr_t id, int signal) {
return (TEMP_FAILURE_RETRY(kill(id, signal)) != -1);
}
runtime/bin/process_patch.dart
+79
View File
@@ -49,6 +49,25 @@ patch class Process {
stdoutEncoding,
stderrEncoding);
}
/* patch */ static ProcessResult runSync(
String executable,
List<String> arguments,
{String workingDirectory,
Map<String, String> environment,
bool includeParentEnvironment: true,
bool runInShell: false,
Encoding stdoutEncoding: Encoding.SYSTEM,
Encoding stderrEncoding: Encoding.SYSTEM}) {
return _runNonInteractiveProcessSync(executable,
arguments,
workingDirectory,
environment,
includeParentEnvironment,
runInShell,
stdoutEncoding,
stderrEncoding);
}
}
@@ -278,6 +297,43 @@ class _ProcessImpl extends NativeFieldWrapperClass1 implements Process {
return completer.future;
}
ProcessResult _runAndWait(Encoding stdoutEncoding,
Encoding stderrEncoding) {
var status = new _ProcessStartStatus();
bool success = _startNative(_path,
_arguments,
_workingDirectory,
_environment,
_stdin._sink._nativeSocket,
_stdout._stream._nativeSocket,
_stderr._stream._nativeSocket,
_exitHandler._nativeSocket,
status);
if (!success) {
throw new ProcessException(_path,
_arguments,
status._errorMessage,
status._errorCode);
}
var result = _wait(
_stdin._sink._nativeSocket,
_stdout._stream._nativeSocket,
_stderr._stream._nativeSocket,
_exitHandler._nativeSocket);
getOutput(output, encoding) {
if (stderrEncoding == null) return output;
return _decodeString(output, encoding);
}
return new _ProcessResult(
result[0],
result[1],
getOutput(result[2], stdoutEncoding),
getOutput(result[3], stderrEncoding));
}
bool _startNative(String path,
List<String> arguments,
String workingDirectory,
@@ -288,6 +344,11 @@ class _ProcessImpl extends NativeFieldWrapperClass1 implements Process {
_NativeSocket exitHandler,
_ProcessStartStatus status) native "Process_Start";
_wait(_NativeSocket stdin,
_NativeSocket stdout,
_NativeSocket stderr,
_NativeSocket exitHandler) native "Process_Wait";
Stream<List<int>> get stdout {
return _stdout;
}
@@ -383,6 +444,24 @@ Future<ProcessResult> _runNonInteractiveProcess(String path,
});
}
ProcessResult _runNonInteractiveProcessSync(
String executable,
List<String> arguments,
String workingDirectory,
Map<String, String> environment,
bool includeParentEnvironment,
bool runInShell,
Encoding stdoutEncoding,
Encoding stderrEncoding) {
var process = new _ProcessImpl(executable,
arguments,
workingDirectory,
environment,
includeParentEnvironment,
runInShell);
return process._runAndWait(stdoutEncoding, stderrEncoding);
}
class _ProcessResult implements ProcessResult {
const _ProcessResult(int this.pid,
runtime/bin/process_win.cc
+195
View File
@@ -642,6 +642,201 @@ int Process::Start(const char* path,
}
class BufferList: public BufferListBase {
public:
BufferList() : read_pending_(true) { }
// Indicate that data has been read into the buffer provided to
// overlapped read.
void DataIsRead(intptr_t size) {
ASSERT(read_pending_ == true);
data_size_ += size;
free_size_ -= size;
ASSERT(free_size_ >= 0);
read_pending_ = false;
}
// The access to the read buffer for overlapped read.
void GetReadBuffer(uint8_t** buffer, intptr_t* size) {
ASSERT(!read_pending_);
if (free_size_ == 0) Allocate();
ASSERT(free_size_ > 0);
ASSERT(free_size_ <= kBufferSize);
*buffer = FreeSpaceAddress();
*size = free_size_;
read_pending_ = true;
}
intptr_t GetDataSize() {
return data_size_;
}
uint8_t* GetFirstDataBuffer() {
ASSERT(head_ != NULL);
ASSERT(head_ == tail_);
ASSERT(data_size_ <= kBufferSize);
return head_->data_;
}
void FreeDataBuffer() {
Free();
}
private:
bool read_pending_;
};
class OverlappedHandle {
public:
void Init(HANDLE handle, HANDLE event) {
handle_ = handle;
event_ = event;
ClearOverlapped();
}
bool HasEvent(HANDLE event) {
return event_ == event;
}
bool Read() {
// Get the data read as a result of a completed overlapped operation.
if (overlapped_.InternalHigh > 0) {
buffer_.DataIsRead(overlapped_.InternalHigh);
} else {
buffer_.DataIsRead(0);
}
// Keep reading until error or pending operation.
while (true) {
ClearOverlapped();
uint8_t* buffer;
intptr_t buffer_size;
buffer_.GetReadBuffer(&buffer, &buffer_size);
BOOL ok = ReadFile(handle_, buffer, buffer_size, NULL, &overlapped_);
if (!ok) return GetLastError() == ERROR_IO_PENDING;
buffer_.DataIsRead(overlapped_.InternalHigh);
}
}
Dart_Handle GetData() {
return buffer_.GetData();
}
intptr_t GetDataSize() {
return buffer_.GetDataSize();
}
uint8_t* GetFirstDataBuffer() {
return buffer_.GetFirstDataBuffer();
}
void FreeDataBuffer() {
return buffer_.FreeDataBuffer();
}
void Close() {
CloseHandle(handle_);
CloseHandle(event_);
handle_ = INVALID_HANDLE_VALUE;
overlapped_.hEvent = INVALID_HANDLE_VALUE;
}
private:
void ClearOverlapped() {
memset(&overlapped_, 0, sizeof(overlapped_));
overlapped_.hEvent = event_;
}
OVERLAPPED overlapped_;
HANDLE handle_;
HANDLE event_;
BufferList buffer_;
DISALLOW_ALLOCATION();
};
bool Process::Wait(intptr_t pid,
intptr_t in,
intptr_t out,
intptr_t err,
intptr_t exit_event,
ProcessResult* result) {
// Close input to the process right away.
reinterpret_cast<FileHandle*>(in)->Close();
// All pipes created to the sub-process support overlapped IO.
FileHandle* stdout_handle = reinterpret_cast<FileHandle*>(out);
ASSERT(stdout_handle->SupportsOverlappedIO());
FileHandle* stderr_handle = reinterpret_cast<FileHandle*>(err);
ASSERT(stderr_handle->SupportsOverlappedIO());
FileHandle* exit_handle = reinterpret_cast<FileHandle*>(exit_event);
ASSERT(exit_handle->SupportsOverlappedIO());
// Create three events for overlapped IO. These are created as already
// signalled to ensure they have read called at least once.
static const int kHandles = 3;
HANDLE events[kHandles];
for (int i = 0; i < kHandles; i++) {
events[i] = CreateEvent(NULL, FALSE, TRUE, NULL);
}
// Setup the structure for handling overlapped IO.
OverlappedHandle oh[kHandles];
oh[0].Init(stdout_handle->handle(), events[0]);
oh[1].Init(stderr_handle->handle(), events[1]);
oh[2].Init(exit_handle->handle(), events[2]);
// Continue until all handles are closed.
int alive = kHandles;
while (alive > 0) {
// Blocking call waiting for events from the child process.
DWORD wait_result = WaitForMultipleObjects(alive, events, FALSE, INFINITE);
// Find the handle signalled.
int index = wait_result - WAIT_OBJECT_0;
for (int i = 0; i < kHandles; i++) {
if (oh[i].HasEvent(events[index])) {
bool ok = oh[i].Read();
if (!ok) {
if (GetLastError() == ERROR_BROKEN_PIPE) {
oh[i].Close();
alive--;
if (index < alive) {
events[index] = events[alive];
}
} else if (err != ERROR_IO_PENDING) {
DWORD e = GetLastError();
oh[0].Close();
oh[1].Close();
oh[2].Close();
SetLastError(e);
return false;
}
}
break;
}
}
}
// All handles closed and all data read.
result->set_stdout_data(oh[0].GetData());
result->set_stderr_data(oh[1].GetData());
// Calculate the exit code.
ASSERT(oh[2].GetDataSize() == 8);
uint32_t exit[2];
memcpy(&exit, oh[2].GetFirstDataBuffer(), sizeof(exit));
oh[2].FreeDataBuffer();
intptr_t exit_code = exit[0];
intptr_t negative = exit[1];
if (negative) exit_code = -exit_code;
result->set_exit_code(exit_code);
return true;
}
bool Process::Kill(intptr_t id, int signal) {
USE(signal); // signal is not used on windows.
HANDLE process_handle;
sdk/lib/_internal/lib/io_patch.dart
+12
View File
@@ -210,6 +210,18 @@ patch class Process {
Encoding stderrEncoding: Encoding.SYSTEM}) {
throw new UnsupportedError("Process.run");
}
patch static ProcessResult runSync(
String executable,
List<String> arguments,
{String workingDirectory,
Map<String, String> environment,
bool includeParentEnvironment: true,
bool runInShell: false,
Encoding stdoutEncoding: Encoding.SYSTEM,
Encoding stderrEncoding: Encoding.SYSTEM}) {
throw new UnsupportedError("Process.runSync");
}
}
patch class InternetAddress {
sdk/lib/io/process.dart
+20
View File
@@ -147,6 +147,26 @@ abstract class Process {
Encoding stdoutEncoding: Encoding.SYSTEM,
Encoding stderrEncoding: Encoding.SYSTEM});
/**
* Starts a process and runs it to completion. This is a synchronous
* call and will block until the child process terminates.
*
* The arguments are the same as for `Process.run`.
*
* Returns a `ProcessResult` with the result of running the process,
* i.e., exit code, standard out and standard in.
*/
external static ProcessResult runSync(
String executable,
List<String> arguments,
{String workingDirectory,
Map<String, String> environment,
bool includeParentEnvironment: true,
bool runInShell: false,
Encoding stdoutEncoding: Encoding.SYSTEM,
Encoding stderrEncoding: Encoding.SYSTEM});
/**
* Returns the standard output stream of the process as a [:Stream:].
*
tests/standalone/io/process_sync_script.dart
+24
View File
@@ -0,0 +1,24 @@
// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
//
// Utility script to generate some output on stdout and stderr.
import "dart:math";
import "dart:io";
main() {
var options = new Options();
var blockCount = int.parse(options.arguments[0]);
var stdoutBlockSize = int.parse(options.arguments[1]);
var stderrBlockSize = int.parse(options.arguments[2]);
var stdoutBlock =
new String.fromCharCodes(new List.filled(stdoutBlockSize, 65));
var stderrBlock =
new String.fromCharCodes(new List.filled(stderrBlockSize, 66));
for (int i = 0; i < blockCount; i++) {
stdout.write(stdoutBlock);
stderr.write(stderrBlock);
}
exit(int.parse(options.arguments[3]));
}
tests/standalone/io/process_sync_test.dart
+63
View File
@@ -0,0 +1,63 @@
// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
import "package:expect/expect.dart";
import 'package:path/path.dart';
import "dart:io";
test(int blockCount,
int stdoutBlockSize,
int stderrBlockSize,
int exitCode,
[int nonWindowsExitCode]) {
// Get the Dart script file that generates output.
var scriptFile = new File(join(dirname(Platform.script),
"process_sync_script.dart"));
var args = [scriptFile.path,
blockCount.toString(),
stdoutBlockSize.toString(),
stderrBlockSize.toString(),
exitCode.toString()];
ProcessResult syncResult = Process.runSync(Platform.executable, args);
Expect.equals(blockCount * stdoutBlockSize, syncResult.stdout.length);
Expect.equals(blockCount * stderrBlockSize, syncResult.stderr.length);
if (Platform.isWindows) {
Expect.equals(exitCode, syncResult.exitCode);
} else {
if (nonWindowsExitCode == null) {
Expect.equals(exitCode, syncResult.exitCode);
} else {
Expect.equals(nonWindowsExitCode, syncResult.exitCode);
}
}
Process.run(Platform.executable, args).then((asyncResult) {
Expect.equals(syncResult.stdout, asyncResult.stdout);
Expect.equals(syncResult.stderr, asyncResult.stderr);
Expect.equals(syncResult.exitCode, asyncResult.exitCode);
});
}
main() {
test(10, 10, 10, 0);
test(10, 100, 10, 0);
test(10, 10, 100, 0);
test(100, 1, 10, 0);
test(100, 10, 1, 0);
test(100, 1, 1, 0);
test(1, 100000, 100000, 0);
// The buffer size used in process.h.
var kBufferSize = 16 * 1024;
test(1, kBufferSize, kBufferSize, 0);
test(1, kBufferSize - 1, kBufferSize + 1, 0);
test(kBufferSize - 1, 1, 1, 0);
test(kBufferSize, 1, 1, 0);
test(kBufferSize + 1, 1, 1, 0);
test(10, 10, 10, 1);
test(10, 10, 10, 255);
test(10, 10, 10, -1, 255);
test(10, 10, 10, -255, 1);
}