goalc: make suggested changes, still can't reconnect properly

common/cross_sockets/XSocketServer.cpp

@@ -17,8 +17,7 @@ XSocketServer::XSocketServer(std::function<bool()> 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() {

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<bool()> 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<char> buffer;
 
   bool kill_accept_thread = false;
   bool server_initialized = false;

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;
       }

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);

goalc/compiler/Compiler.cpp

@@ -64,15 +64,14 @@ Compiler::Compiler(const std::string& user_profile, std::unique_ptr<ReplWrapper>
   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<goos::Object> 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<goos::Object> 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<goos::Object> 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,

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<ReplWrapper> 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<FunctionEnv> compile_top_level_function(const std::string& name,
@@ -73,8 +72,6 @@ class Compiler {
                      std::vector<std::pair<std::string, Replxx::Color>> 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<ReplWrapper> 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<goos::Object> read_from_stdin();
   std::set<std::string> lookup_symbol_infos_starting_with(const std::string& prefix) const;
   std::vector<SymbolInfo>* lookup_exact_name_info(const std::string& name) const;
   bool get_true_or_false(const goos::Object& form, const goos::Object& boolean);

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<std::string> 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) {
-  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();
-}

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> _compiler);
+  std::optional<std::string> read_data();
 
  private:
-  std::shared_ptr<Compiler> compiler = nullptr;
-  char* header_buffer = new char[(int)sizeof(ReplServerHeader)];
+  std::vector<char> header_buffer = std::vector<char>((int)sizeof(ReplServerHeader));
 
   void ping_response();
-  void compile_msg(const std::string_view& msg);
 };

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> compiler;
+    std::unique_ptr<Compiler> 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>();
+      compiler = std::make_unique<Compiler>();
       compiler->run_front_end_on_string(cmd);
       return 0;
     }
-    // Otherwise, run the REPL and such
-    compiler = std::make_shared<Compiler>(username, std::make_unique<ReplWrapper>());
-    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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> lock(compiler_mutex);
+        compiler = std::make_unique<Compiler>(username, std::make_unique<ReplWrapper>());
+        status = ReplStatus::OK;
+      }
+      std::string input_from_stdin = compiler->get_repl_input();
+      if (!input_from_stdin.empty()) {
+        // lock, while we compile
+        std::lock_guard<std::mutex> lock(compiler_mutex);
+        status = compiler->handle_repl_string(input_from_stdin);
       }
     }
   } catch (std::exception& e) {

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('<II', len(form), 10)
+
+num_sent = clientSocket.sendall(header + form.encode())
 
-# this shouldn't be necessary but...this is just a test script
-# might imply something is wrong in the C++ code!
-time.sleep(1000000)