mo2/mob
0
0
Fork 0
mirror of https://github.com/ModOrganizer2/mob.git synced 2026-07-27 14:07:05 -07:00
Code Issues Packages Projects Releases Wiki Activity

split pipe

Browse Source
This commit is contained in:
isanae
2020-12-03 17:44:50 -05:00
parent 5780cce87c
commit ef332f0330
6 changed files with 342 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/pipe.cpp
+278
View File
@@ -0,0 +1,278 @@
#include "pch.h"
#include "pipe.h"
#include "context.h"
#include "process.h"
namespace mob
{
static std::atomic<int> g_next_pipe_id(0);
async_pipe::async_pipe(const context& cx)
: cx_(cx), pending_(false), closed_(true)
{
buffer_ = std::make_unique<char[]>(buffer_size);
std::memset(buffer_.get(), 0, buffer_size);
std::memset(&ov_, 0, sizeof(ov_));
}
bool async_pipe::closed() const
{
return closed_;
}
handle_ptr async_pipe::create_for_stdout()
{
return create(true);
}
handle_ptr async_pipe::create_for_stdin()
{
return create(false);
}
handle_ptr async_pipe::create(bool for_stdout)
{
// creating pipe
handle_ptr out(for_stdout ? create_named_pipe() : create_anonymous_pipe());
if (out.get() == INVALID_HANDLE_VALUE)
return {};
ov_.hEvent = ::CreateEvent(nullptr, TRUE, FALSE, nullptr);
if (ov_.hEvent == NULL)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateEvent failed, {}", error_message(e));
}
event_.reset(ov_.hEvent);
closed_ = false;
return out;
}
std::string_view async_pipe::read(bool finish)
{
std::string_view s;
if (closed_)
return s;
if (pending_)
s = check_pending();
else
s = try_read();
if (finish && s.empty())
{
::CancelIo(stdout_.get());
closed_ = true;
}
return s;
}
std::size_t async_pipe::write(std::string_view s)
{
const DWORD n = static_cast<DWORD>(s.size());
DWORD written = 0;
const auto r = ::WriteFile(stdout_.get(), s.data(), n, &written, nullptr);
if (written >= s.size())
stdout_ = {};
return written;
}
HANDLE async_pipe::create_named_pipe()
{
const auto pipe_id = g_next_pipe_id.fetch_add(1) + 1;
const std::wstring pipe_name =
LR"(\\.\pipe\mob_pipe)" + std::to_wstring(pipe_id);
SECURITY_ATTRIBUTES sa = {};
sa.nLength = sizeof(SECURITY_ATTRIBUTES);
sa.bInheritHandle = TRUE;
handle_ptr pipe;
// creating pipe
{
HANDLE pipe_handle = ::CreateNamedPipeW(
pipe_name.c_str(),
PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED|FILE_FLAG_FIRST_PIPE_INSTANCE,
PIPE_TYPE_BYTE|PIPE_READMODE_BYTE|PIPE_WAIT,
1, buffer_size, buffer_size, process::wait_timeout, &sa);
if (pipe_handle == INVALID_HANDLE_VALUE)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateNamedPipeW failed, {}", error_message(e));
}
pipe.reset(pipe_handle);
}
{
// duplicating the handle to read from it
HANDLE output_read = INVALID_HANDLE_VALUE;
const auto r = DuplicateHandle(
GetCurrentProcess(), pipe.get(), GetCurrentProcess(), &output_read,
0, TRUE, DUPLICATE_SAME_ACCESS);
if (!r)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"DuplicateHandle for pipe failed, {}", error_message(e));
}
stdout_.reset(output_read);
}
// creating handle to pipe which is passed to CreateProcess()
HANDLE output_write = ::CreateFileW(
pipe_name.c_str(), FILE_WRITE_DATA|SYNCHRONIZE, 0,
&sa, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, 0);
if (output_write == INVALID_HANDLE_VALUE)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateFileW for pipe failed, {}", error_message(e));
}
return output_write;
}
HANDLE async_pipe::create_anonymous_pipe()
{
SECURITY_ATTRIBUTES saAttr = {};
saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
saAttr.bInheritHandle = TRUE;
// Create a pipe for the child process's STDIN.
HANDLE read_pipe, write_pipe;
if (!CreatePipe(&read_pipe, &write_pipe, &saAttr, 0))
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreatePipe failed, {}", error_message(e));
}
// Ensure the write handle to the pipe for STDIN is not inherited.
if (!SetHandleInformation(write_pipe, HANDLE_FLAG_INHERIT, 0))
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"SetHandleInformation failed, {}", error_message(e));
}
stdout_.reset(write_pipe);
return read_pipe;
}
std::string_view async_pipe::try_read()
{
DWORD bytes_read = 0;
if (!::ReadFile(stdout_.get(), buffer_.get(), buffer_size, &bytes_read, &ov_))
{
const auto e = GetLastError();
switch (e)
{
case ERROR_IO_PENDING:
{
pending_ = true;
break;
}
case ERROR_BROKEN_PIPE:
{
// broken pipe means the process is finished
closed_ = true;
break;
}
default:
{
cx_.bail_out(context::cmd,
"async_pipe read failed, {}", error_message(e));
break;
}
}
return {};
}
MOB_ASSERT(bytes_read <= buffer_size);
return {buffer_.get(), bytes_read};
}
std::string_view async_pipe::check_pending()
{
DWORD bytes_read = 0;
const auto r = WaitForSingleObject(event_.get(), process::wait_timeout);
if (r == WAIT_FAILED) {
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"WaitForSingleObject in async_pipe failed, {}", error_message(e));
}
if (!::GetOverlappedResult(stdout_.get(), &ov_, &bytes_read, FALSE))
{
const auto e = GetLastError();
switch (e)
{
case ERROR_IO_INCOMPLETE:
{
break;
}
case WAIT_TIMEOUT:
{
break;
}
case ERROR_BROKEN_PIPE:
{
// broken pipe means the process is finished
closed_ = true;
break;
}
default:
{
cx_.bail_out(context::cmd,
"GetOverlappedResult failed in async_pipe, {}",
error_message(e));
break;
}
}
return {};
}
MOB_ASSERT(bytes_read <= buffer_size);
::ResetEvent(event_.get());
pending_ = false;
return {buffer_.get(), bytes_read};
}
} // namespace
src/core/pipe.h
+40
View File
@@ -0,0 +1,40 @@
#pragma once
#include "../utility.h"
namespace mob
{
class async_pipe
{
public:
async_pipe(const context& cx);
handle_ptr create_for_stdout();
std::string_view read(bool finish);
handle_ptr create_for_stdin();
std::size_t write(std::string_view s);
bool closed() const;
private:
static const std::size_t buffer_size = 50'000;
const context& cx_;
handle_ptr stdout_;
handle_ptr event_;
std::unique_ptr<char[]> buffer_;
OVERLAPPED ov_;
bool pending_;
bool closed_;
handle_ptr create(bool for_stdout);
HANDLE create_named_pipe();
HANDLE create_anonymous_pipe();
std::string_view try_read();
std::string_view check_pending();
};
} // namespace
src/core/process.cpp
+7 -270
View File
@@ -3,287 +3,18 @@
#include "conf.h"
#include "context.h"
#include "op.h"
#include "pipe.h"
#include "../net.h"
namespace mob
{
const DWORD wait_timeout = 50;
static std::atomic<int> g_next_pipe_id(0);
HANDLE get_bit_bucket()
{
SECURITY_ATTRIBUTES sa { .nLength = sizeof(sa), .bInheritHandle = TRUE };
return ::CreateFileW(L"NUL", GENERIC_WRITE, 0, &sa, OPEN_EXISTING, 0, 0);
}
async_pipe::async_pipe(const context& cx)
: cx_(cx), pending_(false), closed_(true)
{
buffer_ = std::make_unique<char[]>(buffer_size);
std::memset(buffer_.get(), 0, buffer_size);
std::memset(&ov_, 0, sizeof(ov_));
}
bool async_pipe::closed() const
{
return closed_;
}
handle_ptr async_pipe::create_for_stdout()
{
return create(true);
}
handle_ptr async_pipe::create_for_stdin()
{
return create(false);
}
handle_ptr async_pipe::create(bool for_stdout)
{
// creating pipe
handle_ptr out(for_stdout ? create_named_pipe() : create_anonymous_pipe());
if (out.get() == INVALID_HANDLE_VALUE)
return {};
ov_.hEvent = ::CreateEvent(nullptr, TRUE, FALSE, nullptr);
if (ov_.hEvent == NULL)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateEvent failed, {}", error_message(e));
}
event_.reset(ov_.hEvent);
closed_ = false;
return out;
}
std::string_view async_pipe::read(bool finish)
{
std::string_view s;
if (closed_)
return s;
if (pending_)
s = check_pending();
else
s = try_read();
if (finish && s.empty())
{
::CancelIo(stdout_.get());
closed_ = true;
}
return s;
}
std::size_t async_pipe::write(std::string_view s)
{
const DWORD n = static_cast<DWORD>(s.size());
DWORD written = 0;
const auto r = ::WriteFile(stdout_.get(), s.data(), n, &written, nullptr);
if (written >= s.size())
stdout_ = {};
return written;
}
HANDLE async_pipe::create_named_pipe()
{
const auto pipe_id = g_next_pipe_id.fetch_add(1) + 1;
const std::wstring pipe_name =
LR"(\\.\pipe\mob_pipe)" + std::to_wstring(pipe_id);
SECURITY_ATTRIBUTES sa = {};
sa.nLength = sizeof(SECURITY_ATTRIBUTES);
sa.bInheritHandle = TRUE;
handle_ptr pipe;
// creating pipe
{
HANDLE pipe_handle = ::CreateNamedPipeW(
pipe_name.c_str(),
PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED|FILE_FLAG_FIRST_PIPE_INSTANCE,
PIPE_TYPE_BYTE|PIPE_READMODE_BYTE|PIPE_WAIT,
1, buffer_size, buffer_size, wait_timeout, &sa);
if (pipe_handle == INVALID_HANDLE_VALUE)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateNamedPipeW failed, {}", error_message(e));
}
pipe.reset(pipe_handle);
}
{
// duplicating the handle to read from it
HANDLE output_read = INVALID_HANDLE_VALUE;
const auto r = DuplicateHandle(
GetCurrentProcess(), pipe.get(), GetCurrentProcess(), &output_read,
0, TRUE, DUPLICATE_SAME_ACCESS);
if (!r)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"DuplicateHandle for pipe failed, {}", error_message(e));
}
stdout_.reset(output_read);
}
// creating handle to pipe which is passed to CreateProcess()
HANDLE output_write = ::CreateFileW(
pipe_name.c_str(), FILE_WRITE_DATA|SYNCHRONIZE, 0,
&sa, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, 0);
if (output_write == INVALID_HANDLE_VALUE)
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreateFileW for pipe failed, {}", error_message(e));
}
return output_write;
}
HANDLE async_pipe::create_anonymous_pipe()
{
SECURITY_ATTRIBUTES saAttr = {};
saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
saAttr.bInheritHandle = TRUE;
// Create a pipe for the child process's STDIN.
HANDLE read_pipe, write_pipe;
if (!CreatePipe(&read_pipe, &write_pipe, &saAttr, 0))
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"CreatePipe failed, {}", error_message(e));
}
// Ensure the write handle to the pipe for STDIN is not inherited.
if (!SetHandleInformation(write_pipe, HANDLE_FLAG_INHERIT, 0))
{
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"SetHandleInformation failed, {}", error_message(e));
}
stdout_.reset(write_pipe);
return read_pipe;
}
std::string_view async_pipe::try_read()
{
DWORD bytes_read = 0;
if (!::ReadFile(stdout_.get(), buffer_.get(), buffer_size, &bytes_read, &ov_))
{
const auto e = GetLastError();
switch (e)
{
case ERROR_IO_PENDING:
{
pending_ = true;
break;
}
case ERROR_BROKEN_PIPE:
{
// broken pipe means the process is finished
closed_ = true;
break;
}
default:
{
cx_.bail_out(context::cmd,
"async_pipe read failed, {}", error_message(e));
break;
}
}
return {};
}
MOB_ASSERT(bytes_read <= buffer_size);
return {buffer_.get(), bytes_read};
}
std::string_view async_pipe::check_pending()
{
DWORD bytes_read = 0;
const auto r = WaitForSingleObject(event_.get(), wait_timeout);
if (r == WAIT_FAILED) {
const auto e = GetLastError();
cx_.bail_out(context::cmd,
"WaitForSingleObject in async_pipe failed, {}", error_message(e));
}
if (!::GetOverlappedResult(stdout_.get(), &ov_, &bytes_read, FALSE))
{
const auto e = GetLastError();
switch (e)
{
case ERROR_IO_INCOMPLETE:
{
break;
}
case WAIT_TIMEOUT:
{
break;
}
case ERROR_BROKEN_PIPE:
{
// broken pipe means the process is finished
closed_ = true;
break;
}
default:
{
cx_.bail_out(context::cmd,
"GetOverlappedResult failed in async_pipe, {}",
error_message(e));
break;
}
}
return {};
}
MOB_ASSERT(bytes_read <= buffer_size);
::ResetEvent(event_.get());
pending_ = false;
return {buffer_.get(), bytes_read};
}
process::impl::impl(const impl& i)
: interrupt(i.interrupt.load())
@@ -312,6 +43,12 @@ process::process() :
success_.insert(0);
}
// anchors
process::process(process&&) = default;
process::process(const process&) = default;
process& process::operator=(const process&) = default;
process& process::operator=(process&&) = default;
process::~process()
{
join();
src/core/process.h
+9 -34
View File
@@ -8,40 +8,7 @@ namespace mob
{
class url;
class async_pipe
{
public:
async_pipe(const context& cx);
handle_ptr create_for_stdout();
std::string_view read(bool finish);
handle_ptr create_for_stdin();
std::size_t write(std::string_view s);
bool closed() const;
private:
static const std::size_t buffer_size = 50'000;
const context& cx_;
handle_ptr stdout_;
handle_ptr event_;
std::unique_ptr<char[]> buffer_;
OVERLAPPED ov_;
bool pending_;
bool closed_;
handle_ptr create(bool for_stdout);
HANDLE create_named_pipe();
HANDLE create_anonymous_pipe();
std::string_view try_read();
std::string_view check_pending();
};
class async_pipe;
class encoded_buffer
{
@@ -172,6 +139,8 @@ private:
class process
{
public:
static constexpr DWORD wait_timeout = 50;
enum flags_t
{
noflags = 0x00,
@@ -227,6 +196,12 @@ public:
process();
~process();
// anchors
process(process&&);
process(const process&);
process& operator=(const process&);
process& operator=(process&&);
static process raw(const context& cx, const std::string& cmd);
static process pipe(process p)
vs/mob.vcxproj
+2
View File
@@ -70,6 +70,7 @@
<ClCompile Include="..\src\core\context.cpp" />
<ClCompile Include="..\src\core\env.cpp" />
<ClCompile Include="..\src\core\op.cpp" />
<ClCompile Include="..\src\core\pipe.cpp" />
<ClCompile Include="..\src\core\process.cpp" />
<ClCompile Include="..\src\main.cpp" />
<ClCompile Include="..\src\net.cpp" />
@@ -125,6 +126,7 @@
<ClInclude Include="..\src\core\context.h" />
<ClInclude Include="..\src\core\env.h" />
<ClInclude Include="..\src\core\op.h" />
<ClInclude Include="..\src\core\pipe.h" />
<ClInclude Include="..\src\core\process.h" />
<ClInclude Include="..\src\net.h" />
<ClInclude Include="..\src\pch.h" />
vs/mob.vcxproj.filters
+6
View File
@@ -192,6 +192,9 @@
<ClCompile Include="..\src\cmd\pr.cpp">
<Filter>src\cmd</Filter>
</ClCompile>
<ClCompile Include="..\src\core\pipe.cpp">
<Filter>src\core</Filter>
</ClCompile>
</ItemGroup>
<ItemGroup>
<ClInclude Include="..\src\pch.h">
@@ -251,5 +254,8 @@
<ClInclude Include="..\src\utility\enum.h">
<Filter>src\utility</Filter>
</ClInclude>
<ClInclude Include="..\src\core\pipe.h">
<Filter>src\core</Filter>
</ClInclude>
</ItemGroup>
</Project>