diff --git a/common/cross_sockets/XSocketServer.cpp b/common/cross_sockets/XSocketServer.cpp index 5a70e85af0..0e7dbe92a2 100644 --- a/common/cross_sockets/XSocketServer.cpp +++ b/common/cross_sockets/XSocketServer.cpp @@ -17,8 +17,7 @@ XSocketServer::XSocketServer(std::function shutdown_callback, int _buffer_size) : want_exit_callback(std::move(shutdown_callback)) { tcp_port = _tcp_port; - buffer_size = _buffer_size; - buffer = new char[_buffer_size]; + buffer.resize(_buffer_size); } XSocketServer::~XSocketServer() { @@ -26,20 +25,16 @@ XSocketServer::~XSocketServer() { } void XSocketServer::shutdown_server() { - // Close the listening and accepted socket socket - close_server_socket(); - close_socket(accepted_socket); - - // If the accept thread is still running (nothing ever connected) - // kill it and clean it up + // Cleanup the accept thread if (accept_thread_running) { kill_accept_thread = true; accept_thread.join(); accept_thread_running = false; } - // Cleanup our buffer - delete[] buffer; + // Close the listening and accepted socket socket + close_server_socket(); + close_socket(accepted_socket); } bool XSocketServer::init_server() { @@ -88,8 +83,6 @@ bool XSocketServer::init_server() { } server_initialized = true; - accept_thread_running = true; - kill_accept_thread = false; accept_thread = std::thread(&XSocketServer::accept_thread_func, this); fmt::print("[XSocketServer:{}] awaiting connections\n", tcp_port); return true; @@ -103,26 +96,19 @@ void XSocketServer::close_server_socket() { void XSocketServer::accept_thread_func() { socklen_t l = sizeof(addr); while (!kill_accept_thread) { - accepted_socket = accept_socket(listening_socket, (sockaddr*)&addr, &l); - if (accepted_socket >= 0) { - set_socket_timeout(accepted_socket, 100000); + if (accepted_socket == -1) { + this->accepted_socket = accept_socket(listening_socket, (sockaddr*)&addr, &l); + fmt::print("Accept Socket in XSocketServer: {}\n", this->accepted_socket); + set_socket_timeout(this->accepted_socket, 100000); write_on_accept(); client_connected = true; - return; } + std::this_thread::sleep_for(std::chrono::microseconds(50000)); } } bool XSocketServer::wait_for_connection() { - if (client_connected) { - if (accept_thread_running) { - accept_thread.join(); - accept_thread_running = false; - } - return true; - } else { - return false; - } + return client_connected; } void XSocketServer::lock() { diff --git a/common/cross_sockets/XSocketServer.h b/common/cross_sockets/XSocketServer.h index 3240e93dca..6177993079 100644 --- a/common/cross_sockets/XSocketServer.h +++ b/common/cross_sockets/XSocketServer.h @@ -11,10 +11,12 @@ class XSocketServer { public: static constexpr int DEF_BUFFER_SIZE = 32 * 1024 * 1024; + XSocketServer(const XSocketServer&) = delete; + XSocketServer& operator=(const XSocketServer&) = delete; XSocketServer(std::function shutdown_callback, int _tcp_port, int _buffer_size = DEF_BUFFER_SIZE); - ~XSocketServer(); + virtual ~XSocketServer(); bool init_server(); void shutdown_server(); void close_server_socket(); @@ -25,16 +27,13 @@ class XSocketServer { // Abstract methods -- use-case dependent virtual void write_on_accept() = 0; - virtual void read_data() = 0; - virtual void send_data(void* buf, u16 len) = 0; protected: - int buffer_size; int tcp_port; struct sockaddr_in addr = {}; int listening_socket = -1; int accepted_socket = -1; - char* buffer = nullptr; + std::vector buffer; bool kill_accept_thread = false; bool server_initialized = false; diff --git a/game/system/Deci2Server.cpp b/game/system/Deci2Server.cpp index d073e4e9cd..bfbbab03a2 100644 --- a/game/system/Deci2Server.cpp +++ b/game/system/Deci2Server.cpp @@ -52,15 +52,15 @@ void Deci2Server::read_data() { int got = 0; while (got < desired_size) { - ASSERT(got + desired_size < buffer_size); - auto x = read_from_socket(accepted_socket, buffer + got, desired_size - got); + ASSERT(got + desired_size < buffer.size()); + auto x = read_from_socket(accepted_socket, buffer.data() + got, desired_size - got); if (want_exit_callback()) { return; } got += x > 0 ? x : 0; } - auto* hdr = (Deci2Header*)(buffer); + auto* hdr = (Deci2Header*)(buffer.data()); fprintf(stderr, "[DECI2] Got message: %d %d 0x%x %c -> %c\n", hdr->len, hdr->rsvd, hdr->proto, hdr->src, hdr->dst); @@ -94,7 +94,7 @@ void Deci2Server::read_data() { if (sent_to_program < hdr->rsvd) { // driver.next_recv_size = 0; // driver.next_recv = nullptr; - driver.recv_buffer = buffer + sent_to_program; + driver.recv_buffer = buffer.data() + sent_to_program; driver.available_to_receive = hdr->rsvd - sent_to_program; (driver.handler)(DECI2_READ, driver.available_to_receive, driver.opt); // memcpy(driver.next_recv, buffer + sent_to_program, driver.next_recv_size); @@ -103,7 +103,7 @@ void Deci2Server::read_data() { // receive from network if (hdr->rsvd < hdr->len) { - auto x = read_from_socket(accepted_socket, buffer + hdr->rsvd, hdr->len - hdr->rsvd); + auto x = read_from_socket(accepted_socket, buffer.data() + hdr->rsvd, hdr->len - hdr->rsvd); if (want_exit_callback()) { return; } diff --git a/game/system/Deci2Server.h b/game/system/Deci2Server.h index a92dc1b0cd..67a14adf8c 100644 --- a/game/system/Deci2Server.h +++ b/game/system/Deci2Server.h @@ -12,8 +12,8 @@ class Deci2Server : public XSocketServer { using XSocketServer::XSocketServer; void write_on_accept() override; - void read_data() override; - void send_data(void* buf, u16 len) override; + void read_data(); + void send_data(void* buf, u16 len); void wait_for_protos_ready(); void send_proto_ready(Deci2Driver* drivers, int* driver_count); diff --git a/goalc/compiler/Compiler.cpp b/goalc/compiler/Compiler.cpp index f9663f0220..3ee1caafaa 100644 --- a/goalc/compiler/Compiler.cpp +++ b/goalc/compiler/Compiler.cpp @@ -64,15 +64,14 @@ Compiler::Compiler(const std::string& user_profile, std::unique_ptr setup_goos_forms(); } -void Compiler::lock() { - compiler_mutex.lock(); +Compiler::~Compiler() { + if (m_listener.is_connected()) { + m_listener.send_reset(false); // reset the target + m_listener.disconnect(); + } } -void Compiler::unlock() { - compiler_mutex.unlock(); -} - -std::optional Compiler::read_from_stdin() { +std::string Compiler::get_repl_input() { std::string prompt = fmt::format(fmt::emphasis::bold | fg(fmt::color::cyan), "g > "); if (m_listener.is_connected()) { prompt = fmt::format(fmt::emphasis::bold | fg(fmt::color::lime_green), "gc> "); @@ -82,45 +81,24 @@ std::optional Compiler::read_from_stdin() { } else if (m_debugger.is_attached()) { prompt = fmt::format(fmt::emphasis::bold | fg(fmt::color::red), "gr> "); } - // 1). get a line from the user (READ) - std::optional code = m_goos.reader.read_from_stdin(prompt, *m_repl); - - if (!code) { - return std::nullopt; + std::string prompt_full = "\033[0m" + prompt; + auto str = m_repl->readline(prompt_full); + if (str) { + m_repl->add_to_history(str); + return str; + } else { + return ""; } - return code; } -goos::Object Compiler::read_from_string(const std::string& input) { - return m_goos.reader.read_from_string(input); -} - -ReplStatus Compiler::execute_repl() { - while (!m_want_exit && !m_want_reload) { - auto code = read_from_stdin(); - if (code) { - eval_and_print(code.value()); - } +ReplStatus Compiler::handle_repl_string(const std::string& input) { + if (input.empty()) { + return ReplStatus::OK; } - if (m_listener.is_connected()) { - m_listener.send_reset(false); // reset the target - m_listener.disconnect(); - } - - if (m_want_exit) { - return ReplStatus::WANT_EXIT; - } - - if (m_want_reload) { - return ReplStatus::WANT_RELOAD; - } - - return ReplStatus::OK; -} - -void Compiler::eval_and_print(goos::Object code) { try { + // 1). read + goos::Object code = m_goos.reader.read_from_string(input, true); // 2). compile auto obj_file = compile_object_file("repl", code, m_listener.is_connected()); if (m_settings.debug_print_ir) { @@ -142,10 +120,19 @@ void Compiler::eval_and_print(goos::Object code) { } } } - } catch (std::exception& e) { print_compiler_warning("REPL Error: {}\n", e.what()); } + + if (m_want_exit) { + return ReplStatus::WANT_EXIT; + } + + if (m_want_reload) { + return ReplStatus::WANT_RELOAD; + } + + return ReplStatus::OK; } FileEnv* Compiler::compile_object_file(const std::string& name, diff --git a/goalc/compiler/Compiler.h b/goalc/compiler/Compiler.h index 17469891c7..8a54752e3d 100644 --- a/goalc/compiler/Compiler.h +++ b/goalc/compiler/Compiler.h @@ -29,10 +29,9 @@ enum class ReplStatus { OK, WANT_EXIT, WANT_RELOAD }; class Compiler { public: Compiler(const std::string& user_profile = "#f", std::unique_ptr repl = nullptr); - goos::Object read_from_string(const std::string& input); - void eval_and_print(goos::Object code); - - ReplStatus execute_repl(); + ~Compiler(); + std::string get_repl_input(); + ReplStatus handle_repl_string(const std::string& input); goos::Interpreter& get_goos() { return m_goos; } FileEnv* compile_object_file(const std::string& name, goos::Object code, bool allow_emit); std::unique_ptr compile_top_level_function(const std::string& name, @@ -73,8 +72,6 @@ class Compiler { std::vector> const& user_data); bool knows_object_file(const std::string& name); MakeSystem& make_system() { return m_make; } - void lock(); - void unlock(); private: TypeSystem m_ts; @@ -94,7 +91,6 @@ class Compiler { SymbolInfoMap m_symbol_info; std::unique_ptr m_repl; MakeSystem m_make; - std::mutex compiler_mutex; struct DebugStats { int num_spills = 0; @@ -105,7 +101,6 @@ class Compiler { } m_debug_stats; void setup_goos_forms(); - std::optional read_from_stdin(); std::set lookup_symbol_infos_starting_with(const std::string& prefix) const; std::vector* lookup_exact_name_info(const std::string& name) const; bool get_true_or_false(const goos::Object& form, const goos::Object& boolean); diff --git a/goalc/compiler/nrepl/ReplServer.cpp b/goalc/compiler/nrepl/ReplServer.cpp index 31b46d71d9..8c73f8edec 100644 --- a/goalc/compiler/nrepl/ReplServer.cpp +++ b/goalc/compiler/nrepl/ReplServer.cpp @@ -11,39 +11,61 @@ // // TODO - The server also needs to eventually return the result of the evaluation -// Known Issues: -// - doesn't handle disconnects/reconnects - void ReplServer::write_on_accept() { ping_response(); } -void ReplServer::read_data() { - int desired_size = (int)sizeof(ReplServerHeader); +std::optional ReplServer::read_data() { + if (accepted_socket == -1) { + return std::nullopt; + } int got = 0; - while (got < desired_size) { - ASSERT(got + desired_size < buffer_size); - int sock = accepted_socket; - auto x = read_from_socket(sock, header_buffer + got, desired_size - got); + lock(); + + while (got < header_buffer.size()) { + if (got > header_buffer.size()) { + fmt::print(stderr, "[nREPL]: Bad header, aborting the read. Got :{}, Expected Size: {}", got, + header_buffer.size()); + unlock(); + return std::nullopt; + } + auto x = + read_from_socket(accepted_socket, header_buffer.data() + got, header_buffer.size() - got); if (want_exit_callback()) { - return; + unlock(); + return std::nullopt; + } + if (x == 0 || x == -1) { + accepted_socket = -1; + unlock(); + return std::nullopt; } got += x > 0 ? x : 0; } - auto* header = (ReplServerHeader*)(header_buffer); - - lock(); + auto* header = (ReplServerHeader*)(header_buffer.data()); // get the body of the message - desired_size = header->length; + int expected_size = header->length; got = 0; - while (got < desired_size) { - ASSERT(got + desired_size < buffer_size); - auto x = read_from_socket(accepted_socket, buffer + got, desired_size - got); + while (got < expected_size) { + if (got + expected_size > buffer.size()) { + fmt::print(stderr, + "[nREPL]: Bad message, aborting the read. Got :{}, Expected: {}, Buffer Size: {}", + got, expected_size, buffer.size()); + unlock(); + return std::nullopt; + } + auto x = read_from_socket(accepted_socket, buffer.data() + got, expected_size - got); if (want_exit_callback()) { - return; + unlock(); + return std::nullopt; + } + if (x == 0 || x == -1) { + accepted_socket = -1; + unlock(); + return std::nullopt; } got += x > 0 ? x : 0; } @@ -51,50 +73,23 @@ void ReplServer::read_data() { switch (header->type) { case ReplServerMessageType::PING: ping_response(); - break; + unlock(); + return std::nullopt; case ReplServerMessageType::EVAL: - std::string msg; - msg.assign(buffer, got); - compile_msg(msg); - break; + std::string msg(buffer.data(), header->length); + unlock(); + return std::make_optional(msg); } unlock(); -} - -void ReplServer::send_data(void* buf, u16 len) { - lock(); - if (client_connected) { - int bytes_sent = 0; - while (bytes_sent < len) { - int wrote = write_to_socket(accepted_socket, (char*)(buf) + bytes_sent, len - bytes_sent); - bytes_sent += wrote; - if (!client_connected || want_exit_callback()) { - unlock(); - return; - } - } - } - unlock(); -} - -void ReplServer::set_compiler(std::shared_ptr _compiler) { - compiler = std::move(_compiler); + return std::nullopt; } void ReplServer::ping_response() { std::string ping = fmt::format("Connected to OpenGOAL v{}.{} nREPL!", versions::GOAL_VERSION_MAJOR, versions::GOAL_VERSION_MINOR); lock(); - write_to_socket(accepted_socket, ping.c_str(), ping.size()); + fmt::print("Accept Socket in ReplServer: {}\n", this->accepted_socket); + write_to_socket(this->accepted_socket, ping.c_str(), ping.size()); unlock(); } - -void ReplServer::compile_msg(const std::string_view& msg) { - if (compiler == nullptr) { - return; - } - compiler->lock(); - compiler->eval_and_print(compiler->read_from_string(msg.data())); - compiler->unlock(); -} diff --git a/goalc/compiler/nrepl/ReplServer.h b/goalc/compiler/nrepl/ReplServer.h index 5877e8c303..c8c3ec3798 100644 --- a/goalc/compiler/nrepl/ReplServer.h +++ b/goalc/compiler/nrepl/ReplServer.h @@ -16,15 +16,10 @@ class ReplServer : public XSocketServer { using XSocketServer::XSocketServer; void write_on_accept() override; - void read_data() override; - void send_data(void* buf, u16 len) override; - - void set_compiler(std::shared_ptr _compiler); + std::optional read_data(); private: - std::shared_ptr compiler = nullptr; - char* header_buffer = new char[(int)sizeof(ReplServerHeader)]; + std::vector header_buffer = std::vector((int)sizeof(ReplServerHeader)); void ping_response(); - void compile_msg(const std::string_view& msg); }; diff --git a/goalc/main.cpp b/goalc/main.cpp index 0eb04c3e57..c8fa7c1750 100644 --- a/goalc/main.cpp +++ b/goalc/main.cpp @@ -93,22 +93,25 @@ int main(int argc, char** argv) { std::thread nrepl_thread; // the compiler may throw an exception if it fails to load its standard library. try { - std::shared_ptr compiler; + std::unique_ptr compiler; + std::mutex compiler_mutex; // if a command is provided on the command line, no REPL just run the compiler on it if (!cmd.empty()) { - compiler = std::make_shared(); + compiler = std::make_unique(); compiler->run_front_end_on_string(cmd); return 0; } - // Otherwise, run the REPL and such - compiler = std::make_shared(username, std::make_unique()); - repl_server.set_compiler(compiler); // Start nREPL Server if (repl_server_ok) { nrepl_thread = std::thread([&]() { while (!shutdown_callback()) { if (repl_server.wait_for_connection()) { - repl_server.read_data(); + auto resp = repl_server.read_data(); + if (resp) { + std::lock_guard lock(compiler_mutex); + std::string copy = resp.value(); + status = compiler->handle_repl_string(copy); + } } else { std::this_thread::sleep_for(std::chrono::microseconds(50000)); } @@ -117,16 +120,26 @@ int main(int argc, char** argv) { } // Run automatic forms if applicable if (auto_debug || auto_listen) { - compiler->eval_and_print(compiler->read_from_string("(lt)")); + std::lock_guard lock(compiler_mutex); + status = compiler->handle_repl_string("(lt)"); } if (auto_debug) { - compiler->eval_and_print(compiler->read_from_string("(dbg) (:cont)")); + std::lock_guard lock(compiler_mutex); + status = compiler->handle_repl_string("(dbg) (:cont)"); } // Poll Terminal - while (status == ReplStatus::WANT_RELOAD) { - status = compiler->execute_repl(); + while (status != ReplStatus::WANT_EXIT) { if (status == ReplStatus::WANT_RELOAD) { fmt::print("Reloading compiler...\n"); + std::lock_guard lock(compiler_mutex); + compiler = std::make_unique(username, std::make_unique()); + status = ReplStatus::OK; + } + std::string input_from_stdin = compiler->get_repl_input(); + if (!input_from_stdin.empty()) { + // lock, while we compile + std::lock_guard lock(compiler_mutex); + status = compiler->handle_repl_string(input_from_stdin); } } } catch (std::exception& e) { diff --git a/scripts/nrepl-test.py b/scripts/nrepl-test.py index a0d1a58ac1..21e7887c87 100644 --- a/scripts/nrepl-test.py +++ b/scripts/nrepl-test.py @@ -1,16 +1,15 @@ import socket -import time +import struct clientSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM); clientSocket.connect(("127.0.0.1", 8181)) print(clientSocket) data = clientSocket.recv(1024) print(data.decode()) +print("...?") form = "(repl-help)" -num_sent = clientSocket.send(b'\x0B\x00\x00\x00\x0A\x00\x00\x00' + form.encode()) -print("Sent {} bytes".format(num_sent)) +header = struct.pack('