diff options
author | Eugene Ostroukhov <eostroukhov@google.com> | 2018-05-22 02:59:04 +0300 |
---|---|---|
committer | Anna Henningsen <anna@addaleax.net> | 2018-07-14 00:42:50 +0300 |
commit | 39977db7c015e94d33885249f50e62fa8b1f1bb9 (patch) | |
tree | fb330cb04106d7a030b1599549f5503c3ea086ea /src/inspector | |
parent | 9374a83d6983710844c5436f32c14242ba600a20 (diff) |
inspector: split main thread interface from transport
Workers debugging will require interfacing between the "main" inspector
and per-worker isolate inspectors. This is consistent with what WS
interface does. This change is a refactoring change and does not change
the functionality.
PR-URL: https://github.com/nodejs/node/pull/21182
Fixes: https://github.com/nodejs/node/issues/21725
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Diffstat (limited to 'src/inspector')
-rw-r--r-- | src/inspector/main_thread_interface.cc | 317 | ||||
-rw-r--r-- | src/inspector/main_thread_interface.h | 99 |
2 files changed, 416 insertions, 0 deletions
diff --git a/src/inspector/main_thread_interface.cc b/src/inspector/main_thread_interface.cc new file mode 100644 index 00000000000..da43c95bea5 --- /dev/null +++ b/src/inspector/main_thread_interface.cc @@ -0,0 +1,317 @@ +#include "main_thread_interface.h" + +#include "node_mutex.h" +#include "v8-inspector.h" + +#include <unicode/unistr.h> + +namespace node { +namespace inspector { +namespace { + +using v8_inspector::StringView; +using v8_inspector::StringBuffer; + +template <typename T> +class DeleteRequest : public Request { + public: + explicit DeleteRequest(T* object) : object_(object) {} + void Call() override { + delete object_; + } + + private: + T* object_; +}; + +template <typename Target, typename Arg> +class SingleArgumentFunctionCall : public Request { + public: + using Fn = void (Target::*)(Arg); + + SingleArgumentFunctionCall(Target* target, Fn fn, Arg argument) + : target_(target), + fn_(fn), + arg_(std::move(argument)) {} + + void Call() override { + Apply(target_, fn_, std::move(arg_)); + } + + private: + template <typename Element> + void Apply(Element* target, Fn fn, Arg arg) { + (target->*fn)(std::move(arg)); + } + + Target* target_; + Fn fn_; + Arg arg_; +}; + +class PostMessageRequest : public Request { + public: + PostMessageRequest(InspectorSessionDelegate* delegate, + StringView message) + : delegate_(delegate), + message_(StringBuffer::create(message)) {} + + void Call() override { + delegate_->SendMessageToFrontend(message_->string()); + } + + private: + InspectorSessionDelegate* delegate_; + std::unique_ptr<StringBuffer> message_; +}; + +class DispatchMessagesTask : public v8::Task { + public: + explicit DispatchMessagesTask(MainThreadInterface* thread) + : thread_(thread) {} + + void Run() override { + thread_->DispatchMessages(); + } + + private: + MainThreadInterface* thread_; +}; + +void DisposePairCallback(uv_handle_t* ref) { + using AsyncAndInterface = std::pair<uv_async_t, MainThreadInterface*>; + AsyncAndInterface* pair = node::ContainerOf( + &AsyncAndInterface::first, reinterpret_cast<uv_async_t*>(ref)); + delete pair; +} + +template <typename T> +class AnotherThreadObjectReference { + public: + // We create it on whatever thread, just make sure it gets disposed on the + // proper thread. + AnotherThreadObjectReference(std::shared_ptr<MainThreadHandle> thread, + T* object) + : thread_(thread), object_(object) { + } + AnotherThreadObjectReference(AnotherThreadObjectReference&) = delete; + + ~AnotherThreadObjectReference() { + // Disappearing thread may cause a memory leak + CHECK(thread_->Post( + std::unique_ptr<DeleteRequest<T>>(new DeleteRequest<T>(object_)))); + object_ = nullptr; + } + + template <typename Fn, typename Arg> + void Post(Fn fn, Arg argument) const { + using R = SingleArgumentFunctionCall<T, Arg>; + thread_->Post(std::unique_ptr<R>(new R(object_, fn, std::move(argument)))); + } + + T* get() const { + return object_; + } + + private: + std::shared_ptr<MainThreadHandle> thread_; + T* object_; +}; + +class MainThreadSessionState { + public: + MainThreadSessionState( + std::shared_ptr<MainThreadHandle> thread, + bool prevent_shutdown) : thread_(thread), + prevent_shutdown_(prevent_shutdown) {} + + void Connect(std::unique_ptr<InspectorSessionDelegate> delegate) { + Agent* agent = thread_->GetInspectorAgent(); + if (agent != nullptr) + session_ = agent->Connect(std::move(delegate), prevent_shutdown_); + } + + void Dispatch(std::unique_ptr<StringBuffer> message) { + session_->Dispatch(message->string()); + } + + private: + std::shared_ptr<MainThreadHandle> thread_; + bool prevent_shutdown_; + std::unique_ptr<InspectorSession> session_; +}; + +class CrossThreadInspectorSession : public InspectorSession { + public: + CrossThreadInspectorSession( + int id, + std::shared_ptr<MainThreadHandle> thread, + std::unique_ptr<InspectorSessionDelegate> delegate, + bool prevent_shutdown) + : state_(thread, new MainThreadSessionState(thread, prevent_shutdown)) { + state_.Post(&MainThreadSessionState::Connect, std::move(delegate)); + } + + void Dispatch(const StringView& message) override { + state_.Post(&MainThreadSessionState::Dispatch, + StringBuffer::create(message)); + } + + private: + AnotherThreadObjectReference<MainThreadSessionState> state_; +}; + +class ThreadSafeDelegate : public InspectorSessionDelegate { + public: + ThreadSafeDelegate(std::shared_ptr<MainThreadHandle> thread, + std::unique_ptr<InspectorSessionDelegate> delegate) + : thread_(thread), delegate_(thread, delegate.release()) {} + + void SendMessageToFrontend(const v8_inspector::StringView& message) override { + thread_->Post(std::unique_ptr<Request>( + new PostMessageRequest(delegate_.get(), message))); + } + + private: + std::shared_ptr<MainThreadHandle> thread_; + AnotherThreadObjectReference<InspectorSessionDelegate> delegate_; +}; +} // namespace + + +MainThreadInterface::MainThreadInterface(Agent* agent, uv_loop_t* loop, + v8::Isolate* isolate, + v8::Platform* platform) + : agent_(agent), isolate_(isolate), + platform_(platform) { + main_thread_request_.reset(new AsyncAndInterface(uv_async_t(), this)); + CHECK_EQ(0, uv_async_init(loop, &main_thread_request_->first, + DispatchMessagesAsyncCallback)); + // Inspector uv_async_t should not prevent main loop shutdown. + uv_unref(reinterpret_cast<uv_handle_t*>(&main_thread_request_->first)); +} + +MainThreadInterface::~MainThreadInterface() { + if (handle_) + handle_->Reset(); +} + +// static +void MainThreadInterface::DispatchMessagesAsyncCallback(uv_async_t* async) { + AsyncAndInterface* asyncAndInterface = + node::ContainerOf(&AsyncAndInterface::first, async); + asyncAndInterface->second->DispatchMessages(); +} + +// static +void MainThreadInterface::CloseAsync(AsyncAndInterface* pair) { + uv_close(reinterpret_cast<uv_handle_t*>(&pair->first), DisposePairCallback); +} + +void MainThreadInterface::Post(std::unique_ptr<Request> request) { + Mutex::ScopedLock scoped_lock(requests_lock_); + bool needs_notify = requests_.empty(); + requests_.push_back(std::move(request)); + if (needs_notify) { + CHECK_EQ(0, uv_async_send(&main_thread_request_->first)); + if (isolate_ != nullptr && platform_ != nullptr) { + platform_->CallOnForegroundThread(isolate_, + new DispatchMessagesTask(this)); + isolate_->RequestInterrupt([](v8::Isolate* isolate, void* thread) { + static_cast<MainThreadInterface*>(thread)->DispatchMessages(); + }, this); + } + } + incoming_message_cond_.Broadcast(scoped_lock); +} + +bool MainThreadInterface::WaitForFrontendEvent() { + // We allow DispatchMessages reentry as we enter the pause. This is important + // to support debugging the code invoked by an inspector call, such + // as Runtime.evaluate + dispatching_messages_ = false; + if (dispatching_message_queue_.empty()) { + Mutex::ScopedLock scoped_lock(requests_lock_); + while (requests_.empty()) incoming_message_cond_.Wait(scoped_lock); + } + return true; +} + +void MainThreadInterface::DispatchMessages() { + if (dispatching_messages_) + return; + dispatching_messages_ = true; + bool had_messages = false; + do { + if (dispatching_message_queue_.empty()) { + Mutex::ScopedLock scoped_lock(requests_lock_); + requests_.swap(dispatching_message_queue_); + } + had_messages = !dispatching_message_queue_.empty(); + while (!dispatching_message_queue_.empty()) { + MessageQueue::value_type task; + std::swap(dispatching_message_queue_.front(), task); + dispatching_message_queue_.pop_front(); + task->Call(); + } + } while (had_messages); + dispatching_messages_ = false; +} + +std::shared_ptr<MainThreadHandle> MainThreadInterface::GetHandle() { + if (handle_ == nullptr) + handle_ = std::make_shared<MainThreadHandle>(this); + return handle_; +} + +std::unique_ptr<StringBuffer> Utf8ToStringView(const std::string& message) { + icu::UnicodeString utf16 = icu::UnicodeString::fromUTF8( + icu::StringPiece(message.data(), message.length())); + StringView view(reinterpret_cast<const uint16_t*>(utf16.getBuffer()), + utf16.length()); + return StringBuffer::create(view); +} + +std::unique_ptr<InspectorSession> MainThreadHandle::Connect( + std::unique_ptr<InspectorSessionDelegate> delegate, + bool prevent_shutdown) { + return std::unique_ptr<InspectorSession>( + new CrossThreadInspectorSession(++next_session_id_, + shared_from_this(), + std::move(delegate), + prevent_shutdown)); +} + +bool MainThreadHandle::Post(std::unique_ptr<Request> request) { + Mutex::ScopedLock scoped_lock(block_lock_); + if (!main_thread_) + return false; + main_thread_->Post(std::move(request)); + return true; +} + +void MainThreadHandle::Reset() { + Mutex::ScopedLock scoped_lock(block_lock_); + main_thread_ = nullptr; +} + +Agent* MainThreadHandle::GetInspectorAgent() { + Mutex::ScopedLock scoped_lock(block_lock_); + if (main_thread_ == nullptr) + return nullptr; + return main_thread_->inspector_agent(); +} + +std::unique_ptr<InspectorSessionDelegate> +MainThreadHandle::MakeThreadSafeDelegate( + std::unique_ptr<InspectorSessionDelegate> delegate) { + return std::unique_ptr<InspectorSessionDelegate>( + new ThreadSafeDelegate(shared_from_this(), std::move(delegate))); +} + +bool MainThreadHandle::Expired() { + Mutex::ScopedLock scoped_lock(block_lock_); + return main_thread_ == nullptr; +} +} // namespace inspector +} // namespace node diff --git a/src/inspector/main_thread_interface.h b/src/inspector/main_thread_interface.h new file mode 100644 index 00000000000..75df5ffe809 --- /dev/null +++ b/src/inspector/main_thread_interface.h @@ -0,0 +1,99 @@ +#ifndef SRC_INSPECTOR_MAIN_THREAD_INTERFACE_H_ +#define SRC_INSPECTOR_MAIN_THREAD_INTERFACE_H_ + +#if !HAVE_INSPECTOR +#error("This header can only be used when inspector is enabled") +#endif + +#include "env.h" +#include "inspector_agent.h" +#include "node_mutex.h" + +#include <deque> +#include <memory> +#include <unordered_map> +#include <unordered_set> + +namespace v8_inspector { +class StringBuffer; +class StringView; +} // namespace v8_inspector + +namespace node { +namespace inspector { +class Request { + public: + virtual void Call() = 0; + virtual ~Request() {} +}; + +std::unique_ptr<v8_inspector::StringBuffer> Utf8ToStringView( + const std::string& message); + +using MessageQueue = std::deque<std::unique_ptr<Request>>; +class MainThreadInterface; + +class MainThreadHandle : public std::enable_shared_from_this<MainThreadHandle> { + public: + explicit MainThreadHandle(MainThreadInterface* main_thread) + : main_thread_(main_thread) {} + ~MainThreadHandle() { + CHECK_NULL(main_thread_); // main_thread_ should have called Reset + } + std::unique_ptr<InspectorSession> Connect( + std::unique_ptr<InspectorSessionDelegate> delegate, + bool prevent_shutdown); + bool Post(std::unique_ptr<Request> request); + Agent* GetInspectorAgent(); + std::unique_ptr<InspectorSessionDelegate> MakeThreadSafeDelegate( + std::unique_ptr<InspectorSessionDelegate> delegate); + bool Expired(); + + private: + void Reset(); + + MainThreadInterface* main_thread_; + Mutex block_lock_; + int next_session_id_ = 0; + + friend class MainThreadInterface; +}; + +class MainThreadInterface { + public: + MainThreadInterface(Agent* agent, uv_loop_t*, v8::Isolate* isolate, + v8::Platform* platform); + ~MainThreadInterface(); + + void DispatchMessages(); + void Post(std::unique_ptr<Request> request); + bool WaitForFrontendEvent(); + std::shared_ptr<MainThreadHandle> GetHandle(); + Agent* inspector_agent() { + return agent_; + } + + private: + using AsyncAndInterface = std::pair<uv_async_t, MainThreadInterface*>; + + static void DispatchMessagesAsyncCallback(uv_async_t* async); + static void CloseAsync(AsyncAndInterface*); + + MessageQueue requests_; + Mutex requests_lock_; // requests_ live across threads + // This queue is to maintain the order of the messages for the cases + // when we reenter the DispatchMessages function. + MessageQueue dispatching_message_queue_; + bool dispatching_messages_ = false; + ConditionVariable incoming_message_cond_; + // Used from any thread + Agent* const agent_; + v8::Isolate* const isolate_; + v8::Platform* const platform_; + DeleteFnPtr<AsyncAndInterface, CloseAsync> main_thread_request_; + std::shared_ptr<MainThreadHandle> handle_; +}; + +} // namespace inspector +} // namespace node +#endif // SRC_INSPECTOR_MAIN_THREAD_INTERFACE_H_ |