blob: e4430e60d06cb980caab7d0a1e32381f263246e2 [file] [log] [blame]
// Copyright (c) 2011, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
#include "vm/message.h"
#include <utility>
#include "vm/dart_api_state.h"
#include "vm/dart_entry.h"
#include "vm/json_stream.h"
#include "vm/object.h"
#include "vm/port.h"
namespace dart {
const Dart_Port Message::kIllegalPort = 0;
Message::Message(Dart_Port dest_port,
uint8_t* snapshot,
intptr_t snapshot_length,
MessageFinalizableData* finalizable_data,
Priority priority,
Dart_Port delivery_failure_port)
: dest_port_(dest_port),
delivery_failure_port_(delivery_failure_port),
payload_(snapshot),
snapshot_length_(snapshot_length),
finalizable_data_(finalizable_data),
priority_(priority) {
ASSERT((priority == kNormalPriority) ||
(delivery_failure_port == kIllegalPort));
ASSERT(IsSnapshot());
}
Message::Message(Dart_Port dest_port,
ObjectPtr raw_obj,
Priority priority,
Dart_Port delivery_failure_port)
: dest_port_(dest_port),
delivery_failure_port_(delivery_failure_port),
payload_(raw_obj),
priority_(priority) {
ASSERT(!raw_obj->IsHeapObject() || raw_obj->untag()->InVMIsolateHeap());
ASSERT((priority == kNormalPriority) ||
(delivery_failure_port == kIllegalPort));
ASSERT(IsRaw());
}
Message::Message(Dart_Port dest_port,
PersistentHandle* handle,
Priority priority,
Dart_Port delivery_failure_port)
: dest_port_(dest_port),
delivery_failure_port_(delivery_failure_port),
payload_(handle),
snapshot_length_(kPersistentHandleSnapshotLen),
priority_(priority) {
ASSERT((priority == kNormalPriority) ||
(delivery_failure_port == kIllegalPort));
ASSERT(IsPersistentHandle());
}
Message::~Message() {
ASSERT(delivery_failure_port_ == kIllegalPort);
if (IsSnapshot()) {
free(payload_.snapshot_);
}
delete finalizable_data_;
if (IsPersistentHandle()) {
auto isolate_group = IsolateGroup::Current();
isolate_group->api_state()->FreePersistentHandle(
payload_.persistent_handle_);
}
}
bool Message::RedirectToDeliveryFailurePort() {
if (delivery_failure_port_ == kIllegalPort) {
return false;
}
dest_port_ = delivery_failure_port_;
delivery_failure_port_ = kIllegalPort;
return true;
}
intptr_t Message::Id() const {
// Messages are allocated on the C heap. Use the raw address as the id.
return reinterpret_cast<intptr_t>(this);
}
const char* Message::PriorityAsString(Priority priority) {
switch (priority) {
case kNormalPriority:
return "Normal";
break;
case kOOBPriority:
return "OOB";
break;
default:
UNIMPLEMENTED();
return NULL;
}
}
MessageQueue::MessageQueue() {
head_ = NULL;
tail_ = NULL;
}
MessageQueue::~MessageQueue() {
// Ensure that all pending messages have been released.
Clear();
ASSERT(head_ == NULL);
}
void MessageQueue::Enqueue(std::unique_ptr<Message> msg0, bool before_events) {
// TODO(mdempsky): Use unique_ptr internally?
Message* msg = msg0.release();
// Make sure messages are not reused.
ASSERT(msg->next_ == NULL);
if (head_ == NULL) {
// Only element in the queue.
ASSERT(tail_ == NULL);
head_ = msg;
tail_ = msg;
} else {
ASSERT(tail_ != NULL);
if (!before_events) {
// Append at the tail.
tail_->next_ = msg;
tail_ = msg;
} else {
ASSERT(msg->dest_port() == Message::kIllegalPort);
if (head_->dest_port() != Message::kIllegalPort) {
msg->next_ = head_;
head_ = msg;
} else {
Message* cur = head_;
while (cur->next_ != NULL) {
if (cur->next_->dest_port() != Message::kIllegalPort) {
// Splice in the new message at the break.
msg->next_ = cur->next_;
cur->next_ = msg;
return;
}
cur = cur->next_;
}
// All pending messages are isolate library control messages. Append at
// the tail.
ASSERT(tail_ == cur);
ASSERT(tail_->dest_port() == Message::kIllegalPort);
tail_->next_ = msg;
tail_ = msg;
}
}
}
}
std::unique_ptr<Message> MessageQueue::Dequeue() {
Message* result = head_;
if (result != nullptr) {
head_ = result->next_;
// The following update to tail_ is not strictly needed.
if (head_ == nullptr) {
tail_ = nullptr;
}
#if defined(DEBUG)
result->next_ = result; // Make sure to trigger ASSERT in Enqueue.
#endif // DEBUG
return std::unique_ptr<Message>(result);
}
return nullptr;
}
void MessageQueue::Clear() {
std::unique_ptr<Message> cur(head_);
head_ = nullptr;
tail_ = nullptr;
while (cur != nullptr) {
std::unique_ptr<Message> next(cur->next_);
if (cur->RedirectToDeliveryFailurePort()) {
PortMap::PostMessage(std::move(cur));
}
cur = std::move(next);
}
}
MessageQueue::Iterator::Iterator(const MessageQueue* queue) : next_(NULL) {
Reset(queue);
}
MessageQueue::Iterator::~Iterator() {}
void MessageQueue::Iterator::Reset(const MessageQueue* queue) {
ASSERT(queue != NULL);
next_ = queue->head_;
}
// returns false when there are no more messages left.
bool MessageQueue::Iterator::HasNext() {
return next_ != NULL;
}
// Returns the current message and moves forward.
Message* MessageQueue::Iterator::Next() {
Message* current = next_;
next_ = next_->next_;
return current;
}
intptr_t MessageQueue::Length() const {
MessageQueue::Iterator it(this);
intptr_t length = 0;
while (it.HasNext()) {
it.Next();
length++;
}
return length;
}
Message* MessageQueue::FindMessageById(intptr_t id) {
MessageQueue::Iterator it(this);
while (it.HasNext()) {
Message* current = it.Next();
ASSERT(current != NULL);
if (current->Id() == id) {
return current;
}
}
return NULL;
}
void MessageQueue::PrintJSON(JSONStream* stream) {
#ifndef PRODUCT
JSONArray messages(stream);
Object& msg_handler = Object::Handle();
MessageQueue::Iterator it(this);
intptr_t depth = 0;
while (it.HasNext()) {
Message* current = it.Next();
JSONObject message(&messages);
message.AddProperty("type", "Message");
message.AddPropertyF("name", "Isolate Message (%" Px ")", current->Id());
message.AddPropertyF("messageObjectId", "messages/%" Px "", current->Id());
message.AddProperty("size", current->Size());
message.AddProperty("index", depth++);
message.AddPropertyF("_destinationPort", "%" Pd64 "",
static_cast<int64_t>(current->dest_port()));
message.AddProperty("_priority",
Message::PriorityAsString(current->priority()));
// TODO(johnmccutchan): Move port -> handler map out of Dart and into the
// VM, that way we can lookup the handler without invoking Dart code.
msg_handler = DartLibraryCalls::LookupHandler(current->dest_port());
if (msg_handler.IsClosure()) {
// Grab function from closure.
msg_handler = Closure::Cast(msg_handler).function();
}
if (msg_handler.IsFunction()) {
const Function& function = Function::Cast(msg_handler);
message.AddProperty("handler", function);
const Script& script = Script::Handle(function.script());
if (!script.IsNull()) {
message.AddLocation(script, function.token_pos(),
function.end_token_pos());
}
}
}
#endif // !PRODUCT
}
} // namespace dart