update

Author: andarut

common/dl-utils-lite.cpp

@@ -22,7 +22,6 @@
 #include <time.h>
 #include <unistd.h>
 #include <dirent.h>
-#include <tuple>
 
 #include "common/server/crash-dump.h"
 #include "common/stats/provider.h"
@@ -50,9 +49,9 @@ double dl_time() {
 }
 
 void dl_print_backtrace(void **trace, int trace_size) {
-  std::ignore = write (2, "\n------- Stack Backtrace -------\n", 33);
+  write (2, "\n------- Stack Backtrace -------\n", 33);
   backtrace_symbols_fd (trace, trace_size, 2);
-  std::ignore = write (2, "-------------------------------\n", 32);
+  write (2, "-------------------------------\n", 32);
 }
 
 void dl_print_backtrace() {
@@ -72,7 +71,7 @@ void dl_print_backtrace_gdb() {
     name_buf[res] = 0;
     int child_pid = fork();
     if (child_pid < 0) {
-      std::ignore = write (2, "Can't fork() to run gdb\n", 24);
+      write (2, "Can't fork() to run gdb\n", 24);
       _exit (0);
     }
     if (!child_pid) {
@@ -84,7 +83,7 @@ void dl_print_backtrace_gdb() {
       waitpid (child_pid, nullptr, 0);
     }
   } else {
-    std::ignore = write (2, "can't get name of executable file to pass to gdb\n", 49);
+    write (2, "can't get name of executable file to pass to gdb\n", 49);
   }
 }
 

runtime-light/allocator/allocator.cmake

@@ -1,2 +1 @@
-set(RUNTIME_ALLOCATOR_SRC
-        ${BASE_DIR}/runtime-light/allocator/runtime-light-allocator.cpp)
+set(RUNTIME_ALLOCATOR_SRC allocator/runtime-light-allocator.cpp)

runtime-light/allocator/runtime-light-allocator.cpp

@@ -5,9 +5,8 @@
 #include <cstddef>
 #include <cstring>
 
-#include "runtime-core/runtime-core.h"
+#include "runtime-core/utils/kphp-assert-core.h"
 #include "runtime-light/component/component.h"
-#include "runtime-light/utils/panic.h"
 
 namespace {
 // TODO: make it depend on max chunk size, e.g. MIN_EXTRA_MEM_SIZE = f(MAX_CHUNK_SIZE);
@@ -136,4 +135,4 @@ void *RuntimeAllocator::realloc_global_memory(void *mem, size_t new_size, size_t
 
 void RuntimeAllocator::free_global_memory(void *mem, size_t) noexcept {
   get_platform_context()->allocator.free(mem);
-}
+}

runtime-light/component/component.cmake

@@ -0,0 +1 @@
+prepend(RUNTIME_COMPONENT_SRC component/ component.cpp)

runtime-light/component/component.cpp

@@ -3,64 +3,140 @@
 // Distributed under the GPL v3 License, see LICENSE.notice.txt
 
 #include "runtime-light/component/component.h"
+
+#include <chrono>
+#include <cstdint>
+#include <utility>
+
+#include "runtime-core/utils/kphp-assert-core.h"
 #include "runtime-light/core/globals/php-init-scripts.h"
+#include "runtime-light/header.h"
+#include "runtime-light/scheduler/scheduler.h"
+#include "runtime-light/utils/context.h"
 
-void ComponentState::resume_if_was_rescheduled() {
-  if (poll_status == PollStatus::PollReschedule) {
-    // If component was suspended by please yield and there is no awaitable streams
-    main_thread();
-  }
+void ComponentState::init_script_execution() noexcept {
+  kphp_core_context.init();
+  init_php_scripts_in_each_worker(php_script_mutable_globals_singleton, main_task);
+  scheduler.suspend(std::make_pair(main_task.get_handle(), WaitEvent::Rechedule{}));
 }
 
-bool ComponentState::is_stream_already_being_processed(uint64_t stream_d) {
-  return opened_streams.contains(stream_d);
-}
+void ComponentState::process_platform_updates() noexcept {
+  const auto &platform_ctx{*get_platform_context()};
+
+  for (;;) {
+    // check if platform asked for yield
+    if (static_cast<bool>(platform_ctx.please_yield.load())) { // tell the scheduler that we are about to yield
+      php_debug("platform asked for yield");
+      const auto schedule_status{scheduler.schedule(ScheduleEvent::Yield{})};
+      poll_status = schedule_status == ScheduleStatus::Error ? PollStatus::PollFinishedError : PollStatus::PollReschedule;
+      return;
+    }
 
-void ComponentState::resume_if_wait_stream(uint64_t stream_d, StreamStatus status) {
-  if (is_stream_timer(stream_d)) {
-    process_timer(stream_d);
-  } else {
-    process_stream(stream_d, status);
+    // try taking update from the platform
+    if (uint64_t stream_d{}; static_cast<bool>(platform_ctx.take_update(std::addressof(stream_d)))) {
+      if (opened_streams_.contains(stream_d)) { // update on opened stream
+        php_debug("took update on stream %" PRIu64, stream_d);
+        switch (scheduler.schedule(ScheduleEvent::UpdateOnStream{.stream_d = stream_d})) {
+          case ScheduleStatus::Resumed: { // scheduler's resumed a coroutine waiting for update
+            break;
+          }
+          case ScheduleStatus::Skipped: { // no one is waiting for the event yet, so just save it
+            pending_updates_.insert(stream_d);
+            break;
+          }
+          case ScheduleStatus::Error: { // something bad's happened, stop execution
+            poll_status = PollStatus::PollFinishedError;
+            return;
+          }
+        }
+      } else { // update on incoming stream
+        php_debug("got new incoming stream %" PRIu64, stream_d);
+        if (standard_stream_ != INVALID_PLATFORM_DESCRIPTOR) {
+          php_warning("skip new incoming stream since previous one is not closed");
+          release_stream(stream_d);
+          continue;
+        } // TODO: multiple incoming streams (except for http queries)
+        standard_stream_ = stream_d;
+        incoming_streams_.push_back(stream_d);
+        opened_streams_.insert(stream_d);
+        if (const auto schedule_status{scheduler.schedule(ScheduleEvent::IncomingStream{.stream_d = stream_d})}; schedule_status == ScheduleStatus::Error) {
+          poll_status = PollStatus::PollFinishedError;
+          return;
+        }
+      }
+    } else { // we'are out of updates so let the scheduler do whatever it wants
+      switch (scheduler.schedule(ScheduleEvent::NoEvent{})) {
+        case ScheduleStatus::Resumed: { // scheduler's resumed some coroutine, so let's continue scheduling
+          break;
+        }
+        case ScheduleStatus::Skipped: { // scheduler's done nothing, so it's either scheduled all coroutines or is waiting for events
+          poll_status = scheduler.done() ? PollStatus::PollFinishedOk : PollStatus::PollBlocked;
+          return;
+        }
+        case ScheduleStatus::Error: { // something bad's happened, stop execution
+          poll_status = PollStatus::PollFinishedError;
+          return;
+        }
+      }
+    }
   }
+  // unreachable code
+  poll_status = PollStatus::PollFinishedError;
 }
 
-void ComponentState::process_new_input_stream(uint64_t stream_d) {
-  bool already_pending = std::find(incoming_pending_queries.begin(), incoming_pending_queries.end(), stream_d) != incoming_pending_queries.end();
-  if (!already_pending) {
-    php_debug("got new pending query %lu", stream_d);
-    incoming_pending_queries.push_back(stream_d);
-  }
-  if (wait_incoming_stream) {
-    php_debug("start process pending query %lu", stream_d);
-    main_thread();
+uint64_t ComponentState::take_incoming_stream() noexcept {
+  if (incoming_streams_.empty()) {
+    php_warning("can't take incoming stream cause we don't have them");
+    return INVALID_PLATFORM_DESCRIPTOR;
   }
+  const auto stream_d{incoming_streams_.front()};
+  incoming_streams_.pop_front();
+  php_debug("take incoming stream %" PRIu64, stream_d);
+  return stream_d;
 }
 
-void ComponentState::init_script_execution() {
-  kphp_core_context.init();
-  init_php_scripts_in_each_worker(php_script_mutable_globals_singleton, k_main);
-  main_thread = k_main.get_handle();
+uint64_t ComponentState::open_stream(const string &component_name) noexcept {
+  uint64_t stream_d{};
+  if (const auto open_stream_res{get_platform_context()->open(component_name.size(), component_name.c_str(), std::addressof(stream_d))};
+      open_stream_res != OpenStreamResult::OpenStreamOk) {
+    php_warning("can't open stream to %s", component_name.c_str());
+    return INVALID_PLATFORM_DESCRIPTOR;
+  }
+  opened_streams_.insert(stream_d);
+  php_debug("opened a stream %" PRIu64 " to %s", stream_d, component_name.c_str());
+  return stream_d;
 }
 
-bool ComponentState::is_stream_timer(uint64_t stream_d) {
-  return timer_callbacks.contains(stream_d);
+uint64_t ComponentState::set_timer(std::chrono::nanoseconds duration) noexcept {
+  uint64_t timer_d{};
+  if (const auto set_timer_res{get_platform_context()->set_timer(std::addressof(timer_d), static_cast<uint64_t>(duration.count()))};
+      set_timer_res != SetTimerResult::SetTimerOk) {
+    php_warning("can't set timer for %.9f sec", std::chrono::duration<double>(duration).count());
+    return INVALID_PLATFORM_DESCRIPTOR;
+  }
+  opened_streams_.insert(timer_d);
+  php_debug("set timer %" PRIu64 " for %.9f sec", timer_d, std::chrono::duration<double>(duration).count());
+  return timer_d;
 }
 
-void ComponentState::process_timer(uint64_t stream_d) {
+void ComponentState::release_stream(uint64_t stream_d) noexcept {
+  if (stream_d == standard_stream_) {
+    standard_stream_ = INVALID_PLATFORM_DESCRIPTOR;
+  }
+  opened_streams_.erase(stream_d);
+  pending_updates_.erase(stream_d); // also erase pending updates if exists
   get_platform_context()->free_descriptor(stream_d);
-  timer_callbacks[stream_d]();
-  timer_callbacks.erase(stream_d);
-  opened_streams.erase(stream_d);
+  php_debug("released a stream %" PRIu64, stream_d);
 }
 
-void ComponentState::process_stream(uint64_t stream_d, StreamStatus status) {
-  auto expected_status = opened_streams[stream_d];
-  if ((expected_status == StreamRuntimeStatus::WBlocked && status.write_status != IOBlocked)
-      || (expected_status == StreamRuntimeStatus::RBlocked && status.read_status != IOBlocked)) {
-    php_debug("resume on waited query %lu", stream_d);
-    auto suspend_point = awaiting_coroutines[stream_d];
-    awaiting_coroutines.erase(stream_d);
-    php_assert(awaiting_coroutines.empty());
-    suspend_point();
+void ComponentState::release_all_streams() noexcept {
+  const auto &platform_ctx{*get_platform_context()};
+  standard_stream_ = INVALID_PLATFORM_DESCRIPTOR;
+  for (const auto stream_d : opened_streams_) {
+    platform_ctx.free_descriptor(stream_d);
+    php_debug("released a stream %" PRIu64, stream_d);
   }
+  opened_streams_.clear();
+  pending_updates_.clear();
+  incoming_streams_.clear();
 }

runtime-light/component/component.h

@@ -4,78 +4,87 @@
 
 #pragma once
 
-#include <coroutine>
+#include <chrono>
 #include <csetjmp>
 #include <cstddef>
-#include <functional>
-#include <queue>
+#include <cstdint>
 
 #include "runtime-core/memory-resource/resource_allocator.h"
 #include "runtime-core/memory-resource/unsynchronized_pool_resource.h"
 #include "runtime-core/runtime-core.h"
-
 #include "runtime-light/core/globals/php-script-globals.h"
 #include "runtime-light/coroutine/task.h"
-#include "runtime-light/stdlib/output-control.h"
+#include "runtime-light/header.h"
+#include "runtime-light/scheduler/scheduler.h"
+#include "runtime-light/stdlib/fork/fork-context.h"
+#include "runtime-light/stdlib/output/output-buffer.h"
 #include "runtime-light/stdlib/rpc/rpc-context.h"
-#include "runtime-light/stdlib/superglobals.h"
-#include "runtime-light/streams/streams.h"
-#include "runtime-light/utils/context.h"
+
+constexpr uint64_t INVALID_PLATFORM_DESCRIPTOR = 0;
+
+// Coroutine scheduler type. Change it here if you want to use another scheduler
+using CoroutineScheduler = SimpleCoroutineScheduler;
+static_assert(CoroutineSchedulerConcept<CoroutineScheduler>);
 
 struct ComponentState {
-  template<typename Key, typename Value>
-  using unordered_map = memory_resource::stl::unordered_map<Key, Value, memory_resource::unsynchronized_pool_resource>;
+  template<typename T>
+  using unordered_set = memory_resource::stl::unordered_set<T, memory_resource::unsynchronized_pool_resource>;
+
   template<typename T>
   using deque = memory_resource::stl::deque<T, memory_resource::unsynchronized_pool_resource>;
-  static constexpr auto INIT_RUNTIME_ALLOCATOR_SIZE = static_cast<size_t>(512U * 1024U); // 512KB
 
-  ComponentState()
+  ComponentState() noexcept
     : runtime_allocator(INIT_RUNTIME_ALLOCATOR_SIZE, 0)
+    , scheduler(runtime_allocator.memory_resource)
+    , fork_component_context(runtime_allocator.memory_resource)
     , php_script_mutable_globals_singleton(runtime_allocator.memory_resource)
-    , opened_streams(unordered_map<uint64_t, StreamRuntimeStatus>::allocator_type{runtime_allocator.memory_resource})
-    , awaiting_coroutines(unordered_map<uint64_t, std::coroutine_handle<>>::allocator_type{runtime_allocator.memory_resource})
-    , timer_callbacks(unordered_map<uint64_t, std::function<void()>>::allocator_type{runtime_allocator.memory_resource})
-    , incoming_pending_queries(deque<uint64_t>::allocator_type{runtime_allocator.memory_resource})
-    , rpc_component_context(runtime_allocator.memory_resource) {}
+    , rpc_component_context(runtime_allocator.memory_resource)
+    , incoming_streams_(deque<uint64_t>::allocator_type{runtime_allocator.memory_resource})
+    , opened_streams_(unordered_set<uint64_t>::allocator_type{runtime_allocator.memory_resource})
+    , pending_updates_(unordered_set<uint64_t>::allocator_type{runtime_allocator.memory_resource}) {}
 
   ~ComponentState() = default;
 
-  bool not_finished() const noexcept {
-    return poll_status != PollStatus::PollFinishedOk && poll_status != PollStatus::PollFinishedError;
-  }
-
-  void resume_if_was_rescheduled();
-
-  bool is_stream_already_being_processed(uint64_t stream_d);
-
-  void resume_if_wait_stream(uint64_t stream_d, StreamStatus status);
+  void init_script_execution() noexcept;
+  void process_platform_updates() noexcept;
 
-  void process_new_input_stream(uint64_t stream_d);
-
-  void init_script_execution();
+  bool stream_updated(uint64_t stream_d) const noexcept {
+    return pending_updates_.contains(stream_d);
+  }
+  const unordered_set<uint64_t> &opened_streams() const noexcept {
+    return opened_streams_;
+  }
+  const deque<uint64_t> &incoming_streams() const noexcept {
+    return incoming_streams_;
+  }
+  uint64_t standard_stream() const noexcept {
+    return standard_stream_;
+  }
+  uint64_t take_incoming_stream() noexcept;
+  uint64_t open_stream(const string &) noexcept;
+  uint64_t set_timer(std::chrono::nanoseconds) noexcept;
+  void release_stream(uint64_t) noexcept;
+  void release_all_streams() noexcept;
 
   RuntimeAllocator runtime_allocator;
-  task_t<void> k_main;
-  Response response;
-  PhpScriptMutableGlobals php_script_mutable_globals_singleton;
 
+  CoroutineScheduler scheduler;
+  ForkComponentContext fork_component_context;
   PollStatus poll_status = PollStatus::PollReschedule;
-  uint64_t standard_stream = 0;
-  std::coroutine_handle<> main_thread;
-  bool wait_incoming_stream = false;
 
-  unordered_map<uint64_t, StreamRuntimeStatus> opened_streams; // подумать про необходимость opened_streams. Объединить с awaiting_coroutines
-  unordered_map<uint64_t, std::coroutine_handle<>> awaiting_coroutines;
-  unordered_map<uint64_t, std::function<void()>> timer_callbacks;
-  deque<uint64_t> incoming_pending_queries;
+  Response response;
+  PhpScriptMutableGlobals php_script_mutable_globals_singleton;
 
   KphpCoreContext kphp_core_context;
   RpcComponentContext rpc_component_context;
 
 private:
-  bool is_stream_timer(uint64_t stream_d);
+  task_t<void> main_task;
 
-  void process_timer(uint64_t stream_d);
+  uint64_t standard_stream_{INVALID_PLATFORM_DESCRIPTOR};
+  deque<uint64_t> incoming_streams_;
+  unordered_set<uint64_t> opened_streams_;
+  unordered_set<uint64_t> pending_updates_;
 
-  void process_stream(uint64_t stream_d, StreamStatus status);
+  static constexpr auto INIT_RUNTIME_ALLOCATOR_SIZE = static_cast<size_t>(512U * 1024U); // 512KB
 };

runtime-light/core/core.cmake

@@ -1,7 +1,7 @@
-prepend(RUNTIME_LANGUAGE_SRC ${BASE_DIR}/runtime-light/core/globals/
+prepend(RUNTIME_LANGUAGE_SRC core/globals/
         php-script-globals.cpp)
 
-prepend(RUNTIME_KPHP_CORE_CONTEXT_SRC ${BASE_DIR}/runtime-light/core/kphp-core-impl/
+prepend(RUNTIME_KPHP_CORE_CONTEXT_SRC core/kphp-core-impl/
         kphp-core-context.cpp)
 
 

runtime-light/core/globals/php-init-scripts.h

@@ -4,7 +4,7 @@
 
 #pragma once
 
-#include <cstddef>
+#include "runtime-light/coroutine/task.h"
 
 class PhpScriptMutableGlobals;