Commit 1e9b7d3c authored by jdavidberger's avatar jdavidberger Committed by Brad King

server: Switched to a auto model for handles

parent f43b9219
......@@ -26,7 +26,7 @@ void cmEventBasedConnection::on_alloc_buffer(uv_handle_t* handle,
void cmEventBasedConnection::on_read(uv_stream_t* stream, ssize_t nread,
const uv_buf_t* buf)
{
auto conn = reinterpret_cast<cmEventBasedConnection*>(stream->data);
auto conn = static_cast<cmEventBasedConnection*>(stream->data);
if (conn) {
if (nread >= 0) {
conn->ReadData(std::string(buf->base, buf->base + nread));
......@@ -55,7 +55,7 @@ void cmEventBasedConnection::on_write(uv_write_t* req, int status)
void cmEventBasedConnection::on_new_connection(uv_stream_t* stream, int status)
{
(void)(status);
auto conn = reinterpret_cast<cmEventBasedConnection*>(stream->data);
auto conn = static_cast<cmEventBasedConnection*>(stream->data);
if (conn) {
conn->Connect(stream);
......@@ -76,7 +76,7 @@ void cmEventBasedConnection::WriteData(const std::string& _data)
#endif
auto data = _data;
assert(this->WriteStream);
assert(this->WriteStream.get());
if (BufferStrategy) {
data = BufferStrategy->BufferOutMessage(data);
}
......@@ -87,8 +87,7 @@ void cmEventBasedConnection::WriteData(const std::string& _data)
req->req.data = this;
req->buf = uv_buf_init(new char[ds], static_cast<unsigned int>(ds));
memcpy(req->buf.base, data.c_str(), ds);
uv_write(reinterpret_cast<uv_write_t*>(req),
static_cast<uv_stream_t*>(this->WriteStream), &req->buf, 1,
uv_write(reinterpret_cast<uv_write_t*>(req), this->WriteStream, &req->buf, 1,
on_write);
}
......@@ -156,13 +155,11 @@ bool cmConnection::OnServeStart(std::string* errString)
bool cmEventBasedConnection::OnConnectionShuttingDown()
{
if (this->WriteStream) {
if (this->WriteStream.get()) {
this->WriteStream->data = nullptr;
}
if (this->ReadStream) {
this->ReadStream->data = nullptr;
}
this->ReadStream = nullptr;
this->WriteStream = nullptr;
WriteStream.reset();
return true;
}
......@@ -5,6 +5,7 @@
#include "cmConfigure.h" // IWYU pragma: keep
#include "cmUVHandlePtr.h"
#include "cm_uv.h"
#include <cstddef>
......@@ -107,8 +108,6 @@ public:
bool OnConnectionShuttingDown() override;
virtual void OnDisconnect(int errorCode);
uv_stream_t* ReadStream = nullptr;
uv_stream_t* WriteStream = nullptr;
static void on_close(uv_handle_t* handle);
......@@ -119,6 +118,8 @@ public:
}
protected:
cm::uv_stream_ptr WriteStream;
std::string RawReadBuffer;
std::unique_ptr<cmConnectionBufferStrategy> BufferStrategy;
......
......@@ -2,6 +2,8 @@
file Copyright.txt or https://cmake.org/licensing for details. */
#include "cmPipeConnection.h"
#include <algorithm>
#include "cmServer.h"
cmPipeConnection::cmPipeConnection(const std::string& name,
......@@ -13,39 +15,33 @@ cmPipeConnection::cmPipeConnection(const std::string& name,
void cmPipeConnection::Connect(uv_stream_t* server)
{
if (this->ClientPipe) {
if (this->WriteStream.get()) {
// Accept and close all pipes but the first:
uv_pipe_t* rejectPipe = new uv_pipe_t();
cm::uv_pipe_ptr rejectPipe;
rejectPipe.init(*this->Server->GetLoop(), 0);
uv_accept(server, rejectPipe);
uv_pipe_init(this->Server->GetLoop(), rejectPipe, 0);
uv_accept(server, reinterpret_cast<uv_stream_t*>(rejectPipe));
uv_close(reinterpret_cast<uv_handle_t*>(rejectPipe),
&on_close_delete<uv_pipe_t>);
return;
}
this->ClientPipe = new uv_pipe_t();
uv_pipe_init(this->Server->GetLoop(), this->ClientPipe, 0);
this->ClientPipe->data = static_cast<cmEventBasedConnection*>(this);
auto client = reinterpret_cast<uv_stream_t*>(this->ClientPipe);
if (uv_accept(server, client) != 0) {
uv_close(reinterpret_cast<uv_handle_t*>(client),
&on_close_delete<uv_pipe_t>);
this->ClientPipe = nullptr;
cm::uv_pipe_ptr ClientPipe;
ClientPipe.init(*this->Server->GetLoop(), 0,
static_cast<cmEventBasedConnection*>(this));
if (uv_accept(server, ClientPipe) != 0) {
return;
}
this->ReadStream = client;
this->WriteStream = client;
uv_read_start(this->ReadStream, on_alloc_buffer, on_read);
uv_read_start(ClientPipe, on_alloc_buffer, on_read);
WriteStream = std::move(ClientPipe);
Server->OnConnected(this);
}
bool cmPipeConnection::OnServeStart(std::string* errorMessage)
{
this->ServerPipe = new uv_pipe_t();
uv_pipe_init(this->Server->GetLoop(), this->ServerPipe, 0);
this->ServerPipe->data = static_cast<cmEventBasedConnection*>(this);
this->ServerPipe.init(*this->Server->GetLoop(), 0,
static_cast<cmEventBasedConnection*>(this));
int r;
if ((r = uv_pipe_bind(this->ServerPipe, this->PipeName.c_str())) != 0) {
......@@ -53,8 +49,8 @@ bool cmPipeConnection::OnServeStart(std::string* errorMessage)
": " + uv_err_name(r);
return false;
}
auto serverStream = reinterpret_cast<uv_stream_t*>(this->ServerPipe);
if ((r = uv_listen(serverStream, 1, on_new_connection)) != 0) {
if ((r = uv_listen(this->ServerPipe, 1, on_new_connection)) != 0) {
*errorMessage = std::string("Internal Error listening on ") +
this->PipeName + ": " + uv_err_name(r);
return false;
......@@ -65,18 +61,11 @@ bool cmPipeConnection::OnServeStart(std::string* errorMessage)
bool cmPipeConnection::OnConnectionShuttingDown()
{
if (this->ClientPipe) {
uv_close(reinterpret_cast<uv_handle_t*>(this->ClientPipe),
&on_close_delete<uv_pipe_t>);
if (this->WriteStream.get()) {
this->WriteStream->data = nullptr;
}
uv_close(reinterpret_cast<uv_handle_t*>(this->ServerPipe),
&on_close_delete<uv_pipe_t>);
this->ClientPipe = nullptr;
this->ServerPipe = nullptr;
this->WriteStream = nullptr;
this->ReadStream = nullptr;
this->ServerPipe.reset();
return cmEventBasedConnection::OnConnectionShuttingDown();
}
......@@ -4,6 +4,7 @@
#include "cmConfigure.h" // IWYU pragma: keep
#include "cmUVHandlePtr.h"
#include <string>
#include "cmConnection.h"
......@@ -23,6 +24,5 @@ public:
private:
const std::string PipeName;
uv_pipe_t* ServerPipe = nullptr;
uv_pipe_t* ClientPipe = nullptr;
cm::uv_pipe_ptr ServerPipe;
};
......@@ -22,13 +22,14 @@
void on_signal(uv_signal_t* signal, int signum)
{
auto conn = reinterpret_cast<cmServerBase*>(signal->data);
auto conn = static_cast<cmServerBase*>(signal->data);
conn->OnSignal(signum);
}
static void on_walk_to_shutdown(uv_handle_t* handle, void* arg)
{
(void)arg;
assert(uv_is_closing(handle));
if (!uv_is_closing(handle)) {
uv_close(handle, &cmEventBasedConnection::on_close);
}
......@@ -58,6 +59,8 @@ cmServer::cmServer(cmConnection* conn, bool supportExperimental)
cmServer::~cmServer()
{
Close();
for (cmServerProtocol* p : this->SupportedProtocols) {
delete p;
}
......@@ -409,7 +412,7 @@ void cmServer::StartShutDown()
static void __start_thread(void* arg)
{
auto server = reinterpret_cast<cmServerBase*>(arg);
auto server = static_cast<cmServerBase*>(arg);
std::string error;
bool success = server->Serve(&error);
if (!success || error.empty() == false) {
......@@ -417,22 +420,19 @@ static void __start_thread(void* arg)
}
}
static void __shutdownThread(uv_async_t* arg)
{
auto server = reinterpret_cast<cmServerBase*>(arg->data);
on_walk_to_shutdown(reinterpret_cast<uv_handle_t*>(arg), nullptr);
server->StartShutDown();
}
bool cmServerBase::StartServeThread()
{
ServeThreadRunning = true;
uv_async_init(&Loop, &this->ShutdownSignal, __shutdownThread);
this->ShutdownSignal.data = this;
uv_thread_create(&ServeThread, __start_thread, this);
return true;
}
static void __shutdownThread(uv_async_t* arg)
{
auto server = static_cast<cmServerBase*>(arg->data);
server->StartShutDown();
}
bool cmServerBase::Serve(std::string* errorMessage)
{
#ifndef NDEBUG
......@@ -443,14 +443,13 @@ bool cmServerBase::Serve(std::string* errorMessage)
errorMessage->clear();
uv_signal_init(&Loop, &this->SIGINTHandler);
uv_signal_init(&Loop, &this->SIGHUPHandler);
ShutdownSignal.init(Loop, __shutdownThread, this);
this->SIGINTHandler.data = this;
this->SIGHUPHandler.data = this;
SIGINTHandler.init(Loop, this);
SIGHUPHandler.init(Loop, this);
uv_signal_start(&this->SIGINTHandler, &on_signal, SIGINT);
uv_signal_start(&this->SIGHUPHandler, &on_signal, SIGHUP);
SIGINTHandler.start(&on_signal, SIGINT);
SIGHUPHandler.start(&on_signal, SIGHUP);
OnServeStart();
......@@ -473,7 +472,6 @@ bool cmServerBase::Serve(std::string* errorMessage)
return false;
}
ServeThreadRunning = false;
return true;
}
......@@ -487,15 +485,9 @@ void cmServerBase::OnServeStart()
void cmServerBase::StartShutDown()
{
if (!uv_is_closing(
reinterpret_cast<const uv_handle_t*>(&this->SIGINTHandler))) {
uv_signal_stop(&this->SIGINTHandler);
}
if (!uv_is_closing(
reinterpret_cast<const uv_handle_t*>(&this->SIGHUPHandler))) {
uv_signal_stop(&this->SIGHUPHandler);
}
ShutdownSignal.reset();
SIGINTHandler.reset();
SIGHUPHandler.reset();
{
cm::unique_lock<cm::shared_mutex> lock(ConnectionsMutex);
......@@ -519,20 +511,27 @@ cmServerBase::cmServerBase(cmConnection* connection)
{
auto err = uv_loop_init(&Loop);
(void)err;
Loop.data = this;
assert(err == 0);
AddNewConnection(connection);
}
cmServerBase::~cmServerBase()
void cmServerBase::Close()
{
if (Loop.data) {
if (ServeThreadRunning) {
this->ShutdownSignal.send();
uv_thread_join(&ServeThread);
}
if (ServeThreadRunning) {
uv_async_send(&this->ShutdownSignal);
uv_thread_join(&ServeThread);
uv_loop_close(&Loop);
Loop.data = nullptr;
}
uv_loop_close(&Loop);
}
cmServerBase::~cmServerBase()
{
Close();
}
void cmServerBase::AddNewConnection(cmConnection* ownedConnection)
......@@ -562,6 +561,6 @@ void cmServerBase::OnDisconnect(cmConnection* pConnection)
}
if (Connections.empty()) {
StartShutDown();
this->ShutdownSignal.send();
}
}
......@@ -8,6 +8,8 @@
#include "cm_thread.hxx"
#include "cm_uv.h"
#include "cmUVHandlePtr.h"
#include <memory> // IWYU pragma: keep
#include <string>
#include <vector>
......@@ -58,7 +60,7 @@ public:
virtual bool OnSignal(int signum);
uv_loop_t* GetLoop();
void Close();
void OnDisconnect(cmConnection* pConnection);
protected:
......@@ -67,7 +69,7 @@ protected:
bool ServeThreadRunning = false;
uv_thread_t ServeThread;
uv_async_t ShutdownSignal;
cm::uv_async_ptr ShutdownSignal;
#ifndef NDEBUG
public:
// When the server starts it will mark down it's current thread ID,
......@@ -80,8 +82,8 @@ protected:
uv_loop_t Loop;
uv_signal_t SIGINTHandler;
uv_signal_t SIGHUPHandler;
cm::uv_signal_ptr SIGINTHandler;
cm::uv_signal_ptr SIGHUPHandler;
};
class cmServer : public cmServerBase
......
......@@ -5,6 +5,9 @@
#include "cmConfigure.h"
#include "cmServer.h"
#include "cmServerDictionary.h"
#include "cm_uv.h"
#include <algorithm>
#ifdef _WIN32
#include "io.h"
#else
......@@ -18,36 +21,34 @@ cmStdIoConnection::cmStdIoConnection(
{
}
void cmStdIoConnection::SetupStream(uv_stream_t*& stream, int file_id)
cm::uv_stream_ptr cmStdIoConnection::SetupStream(int file_id)
{
assert(stream == nullptr);
switch (uv_guess_handle(file_id)) {
case UV_TTY: {
auto tty = new uv_tty_t();
uv_tty_init(this->Server->GetLoop(), tty, file_id, file_id == 0);
cm::uv_tty_ptr tty;
tty.init(*this->Server->GetLoop(), file_id, file_id == 0,
static_cast<cmEventBasedConnection*>(this));
uv_tty_set_mode(tty, UV_TTY_MODE_NORMAL);
stream = reinterpret_cast<uv_stream_t*>(tty);
break;
return std::move(tty);
}
case UV_FILE:
if (file_id == 0) {
return;
return nullptr;
}
// Intentional fallthrough; stdin can _not_ be treated as a named
// pipe, however stdout can be.
CM_FALLTHROUGH;
case UV_NAMED_PIPE: {
auto pipe = new uv_pipe_t();
uv_pipe_init(this->Server->GetLoop(), pipe, 0);
cm::uv_pipe_ptr pipe;
pipe.init(*this->Server->GetLoop(), 0,
static_cast<cmEventBasedConnection*>(this));
uv_pipe_open(pipe, file_id);
stream = reinterpret_cast<uv_stream_t*>(pipe);
break;
return std::move(pipe);
}
default:
assert(false && "Unable to determine stream type");
return;
return nullptr;
}
stream->data = static_cast<cmEventBasedConnection*>(this);
}
void cmStdIoConnection::SetServer(cmServerBase* s)
......@@ -57,14 +58,14 @@ void cmStdIoConnection::SetServer(cmServerBase* s)
return;
}
SetupStream(this->ReadStream, 0);
SetupStream(this->WriteStream, 1);
this->ReadStream = SetupStream(0);
this->WriteStream = SetupStream(1);
}
void shutdown_connection(uv_prepare_t* prepare)
{
cmStdIoConnection* connection =
reinterpret_cast<cmStdIoConnection*>(prepare->data);
static_cast<cmStdIoConnection*>(prepare->data);
if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(prepare))) {
uv_close(reinterpret_cast<uv_handle_t*>(prepare),
......@@ -76,7 +77,7 @@ void shutdown_connection(uv_prepare_t* prepare)
bool cmStdIoConnection::OnServeStart(std::string* pString)
{
Server->OnConnected(this);
if (this->ReadStream) {
if (this->ReadStream.get()) {
uv_read_start(this->ReadStream, on_alloc_buffer, on_read);
} else if (uv_guess_handle(0) == UV_FILE) {
char buffer[1024];
......@@ -94,44 +95,14 @@ bool cmStdIoConnection::OnServeStart(std::string* pString)
return cmConnection::OnServeStart(pString);
}
void cmStdIoConnection::ShutdownStream(uv_stream_t*& stream)
{
if (!stream) {
return;
}
switch (stream->type) {
case UV_TTY: {
assert(!uv_is_closing(reinterpret_cast<uv_handle_t*>(stream)));
if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(stream))) {
uv_close(reinterpret_cast<uv_handle_t*>(stream),
&on_close_delete<uv_tty_t>);
}
break;
}
case UV_FILE:
case UV_NAMED_PIPE: {
assert(!uv_is_closing(reinterpret_cast<uv_handle_t*>(stream)));
if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(stream))) {
uv_close(reinterpret_cast<uv_handle_t*>(stream),
&on_close_delete<uv_pipe_t>);
}
break;
}
default:
assert(false && "Unable to determine stream type");
}
stream = nullptr;
}
bool cmStdIoConnection::OnConnectionShuttingDown()
{
if (ReadStream) {
if (ReadStream.get()) {
uv_read_stop(ReadStream);
ReadStream->data = nullptr;
}
ShutdownStream(ReadStream);
ShutdownStream(WriteStream);
this->ReadStream.reset();
cmEventBasedConnection::OnConnectionShuttingDown();
......
......@@ -8,7 +8,7 @@
#include "cmConnection.h"
#include "cmPipeConnection.h"
#include "cm_uv.h"
#include "cmUVHandlePtr.h"
class cmServerBase;
......@@ -46,8 +46,8 @@ public:
bool OnServeStart(std::string* pString) override;
private:
void SetupStream(uv_stream_t*& stream, int file_id);
void ShutdownStream(uv_stream_t*& stream);
cm::uv_stream_ptr SetupStream(int file_id);
cm::uv_stream_ptr ReadStream;
};
/***
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment