runtime/vm/port.cc - sdk.git - Git at Google

 // Copyright (c) 2012, 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/port.h"

 #include "platform/utils.h"
 #include "vm/dart_api_impl.h"
 #include "vm/dart_entry.h"
 #include "vm/isolate.h"
 #include "vm/lockers.h"
 #include "vm/message_handler.h"
 #include "vm/os_thread.h"

 namespace dart {

 Mutex* PortMap::mutex_ = NULL;
 PortMap::Entry* PortMap::map_ = NULL;
 MessageHandler* PortMap::deleted_entry_ = reinterpret_cast<MessageHandler*>(1);
 intptr_t PortMap::capacity_ = 0;
 intptr_t PortMap::used_ = 0;
 intptr_t PortMap::deleted_ = 0;
 Random* PortMap::prng_ = NULL;

 intptr_t PortMap::FindPort(Dart_Port port) {
   // ILLEGAL_PORT (0) is used as a sentinel value in Entry.port. The loop below
   // could return the index to a deleted port when we are searching for
   // port id ILLEGAL_PORT. Return -1 immediately to indicate the port
   // does not exist.
   if (port == ILLEGAL_PORT) {
     return -1;
   }
   ASSERT(port != ILLEGAL_PORT);
   intptr_t index = port % capacity_;
   intptr_t start_index = index;
   Entry entry = map_[index];
   while (entry.handler != NULL) {
     if (entry.port == port) {
       return index;
     }
     index = (index + 1) % capacity_;
     // Prevent endless loops.
     ASSERT(index != start_index);
     entry = map_[index];
   }
   return -1;
 }

 void PortMap::Rehash(intptr_t new_capacity) {
   Entry* new_ports = new Entry[new_capacity];
   memset(new_ports, 0, new_capacity * sizeof(Entry));

   for (intptr_t i = 0; i < capacity_; i++) {
     Entry entry = map_[i];
     // Skip free and deleted entries.
     if (entry.port != 0) {
       intptr_t new_index = entry.port % new_capacity;
       while (new_ports[new_index].port != 0) {
         new_index = (new_index + 1) % new_capacity;
       }
       new_ports[new_index] = entry;
     }
   }
   delete[] map_;
   map_ = new_ports;
   capacity_ = new_capacity;
   deleted_ = 0;
 }

 const char* PortMap::PortStateString(PortState kind) {
   switch (kind) {
     case kNewPort:
       return "new";
     case kLivePort:
       return "live";
     case kControlPort:
       return "control";
     default:
       UNREACHABLE();
       return "UNKNOWN";
   }
 }

 Dart_Port PortMap::AllocatePort() {
   const Dart_Port kMASK = 0x3fffffff;
   Dart_Port result = prng_->NextUInt32() & kMASK;

   // Keep getting new values while we have an illegal port number or the port
   // number is already in use.
   while ((result == 0) || (FindPort(result) >= 0)) {
     result = prng_->NextUInt32() & kMASK;
   }

   ASSERT(result != 0);
   ASSERT(FindPort(result) < 0);
   return result;
 }

 void PortMap::SetPortState(Dart_Port port, PortState state) {
   MutexLocker ml(mutex_);
   intptr_t index = FindPort(port);
   ASSERT(index >= 0);
   PortState old_state = map_[index].state;
   ASSERT(old_state == kNewPort);
   map_[index].state = state;
   if (state == kLivePort) {
     map_[index].handler->increment_live_ports();
   }
   if (FLAG_trace_isolates) {
     OS::Print(
         "[^] Port (%s) -> (%s): \n"
         "\thandler:    %s\n"
         "\tport:       %" Pd64 "\n",
         PortStateString(old_state), PortStateString(state),
         map_[index].handler->name(), port);
   }
 }

 void PortMap::MaintainInvariants() {
   intptr_t empty = capacity_ - used_ - deleted_;
   if (used_ > ((capacity_ / 4) * 3)) {
     // Grow the port map.
     Rehash(capacity_ * 2);
   } else if (empty < deleted_) {
     // Rehash without growing the table to flush the deleted slots out of the
     // map.
     Rehash(capacity_);
   }
 }

 Dart_Port PortMap::CreatePort(MessageHandler* handler) {
   ASSERT(handler != NULL);
   MutexLocker ml(mutex_);
 #if defined(DEBUG)
   handler->CheckAccess();
 #endif

   Entry entry;
   entry.port = AllocatePort();
   entry.handler = handler;
   entry.state = kNewPort;

   // Search for the first unused slot. Make use of the knowledge that here is
   // currently no port with this id in the port map.
   ASSERT(FindPort(entry.port) < 0);
   intptr_t index = entry.port % capacity_;
   Entry cur = map_[index];
   // Stop the search at the first found unused (free or deleted) slot.
   while (cur.port != 0) {
     index = (index + 1) % capacity_;
     cur = map_[index];
   }

   // Insert the newly created port at the index.
   ASSERT(index >= 0);
   ASSERT(index < capacity_);
   ASSERT(map_[index].port == 0);
   ASSERT((map_[index].handler == NULL) ||
          (map_[index].handler == deleted_entry_));
   if (map_[index].handler == deleted_entry_) {
     // Consuming a deleted entry.
     deleted_--;
   }
   map_[index] = entry;

   // Increment number of used slots and grow if necessary.
   used_++;
   MaintainInvariants();

   if (FLAG_trace_isolates) {
     OS::Print(
         "[+] Opening port: \n"
         "\thandler:    %s\n"
         "\tport:       %" Pd64 "\n",
         handler->name(), entry.port);
   }

   return entry.port;
 }

 bool PortMap::ClosePort(Dart_Port port) {
   MessageHandler* handler = NULL;
   {
     MutexLocker ml(mutex_);
     intptr_t index = FindPort(port);
     if (index < 0) {
       return false;
     }
     ASSERT(index < capacity_);
     ASSERT(map_[index].port != 0);
     ASSERT(map_[index].handler != deleted_entry_);
     ASSERT(map_[index].handler != NULL);

     handler = map_[index].handler;
 #if defined(DEBUG)
     handler->CheckAccess();
 #endif
     // Before releasing the lock mark the slot in the map as deleted. This makes
     // it possible to release the port map lock before flushing all of its
     // pending messages below.
     map_[index].port = 0;
     map_[index].handler = deleted_entry_;
     if (map_[index].state == kLivePort) {
       handler->decrement_live_ports();
     }

     used_--;
     deleted_++;
     MaintainInvariants();
   }
   handler->ClosePort(port);
   if (!handler->HasLivePorts() && handler->OwnedByPortMap()) {
     // Delete handler as soon as it isn't busy with a task.
     handler->RequestDeletion();
   }
   return true;
 }

 void PortMap::ClosePorts(MessageHandler* handler) {
   {
     MutexLocker ml(mutex_);
     for (intptr_t i = 0; i < capacity_; i++) {
       if (map_[i].handler == handler) {
         // Mark the slot as deleted.
         map_[i].port = 0;
         map_[i].handler = deleted_entry_;
         if (map_[i].state == kLivePort) {
           handler->decrement_live_ports();
         }
         used_--;
         deleted_++;
       }
     }
     MaintainInvariants();
   }
   handler->CloseAllPorts();
 }

 bool PortMap::PostMessage(Message* message) {
   MutexLocker ml(mutex_);
   intptr_t index = FindPort(message->dest_port());
   if (index < 0) {
     delete message;
     return false;
   }
   ASSERT(index >= 0);
   ASSERT(index < capacity_);
   MessageHandler* handler = map_[index].handler;
   ASSERT(map_[index].port != 0);
   ASSERT((handler != NULL) && (handler != deleted_entry_));
   handler->PostMessage(message);
   return true;
 }

 bool PortMap::IsLocalPort(Dart_Port id) {
   MutexLocker ml(mutex_);
   intptr_t index = FindPort(id);
   if (index < 0) {
     // Port does not exist.
     return false;
   }

   MessageHandler* handler = map_[index].handler;
   return handler->IsCurrentIsolate();
 }

 Isolate* PortMap::GetIsolate(Dart_Port id) {
   MutexLocker ml(mutex_);
   intptr_t index = FindPort(id);
   if (index < 0) {
     // Port does not exist.
     return NULL;
   }

   MessageHandler* handler = map_[index].handler;
   return handler->isolate();
 }

 void PortMap::InitOnce() {
   mutex_ = new Mutex();
   prng_ = new Random();

   static const intptr_t kInitialCapacity = 8;
   // TODO(iposva): Verify whether we want to keep exponentially growing.
   ASSERT(Utils::IsPowerOfTwo(kInitialCapacity));
   map_ = new Entry[kInitialCapacity];
   memset(map_, 0, kInitialCapacity * sizeof(Entry));
   capacity_ = kInitialCapacity;
   used_ = 0;
   deleted_ = 0;
 }

 void PortMap::PrintPortsForMessageHandler(MessageHandler* handler,
                                           JSONStream* stream) {
 #ifndef PRODUCT
   if (!FLAG_support_service) {
     return;
   }
   JSONObject jsobj(stream);
   jsobj.AddProperty("type", "_Ports");
   Object& msg_handler = Object::Handle();
   {
     JSONArray ports(&jsobj, "ports");
     SafepointMutexLocker ml(mutex_);
     for (intptr_t i = 0; i < capacity_; i++) {
       if (map_[i].handler == handler) {
         if (map_[i].state == kLivePort) {
           JSONObject port(&ports);
           port.AddProperty("type", "_Port");
           port.AddPropertyF("name", "Isolate Port (%" Pd64 ")", map_[i].port);
           msg_handler = DartLibraryCalls::LookupHandler(map_[i].port);
           port.AddProperty("handler", msg_handler);
         }
       }
     }
   }
 #endif
 }

 void PortMap::DebugDumpForMessageHandler(MessageHandler* handler) {
   SafepointMutexLocker ml(mutex_);
   Object& msg_handler = Object::Handle();
   for (intptr_t i = 0; i < capacity_; i++) {
     if (map_[i].handler == handler) {
       if (map_[i].state == kLivePort) {
         OS::Print("Live Port = %" Pd64 "\n", map_[i].port);
         msg_handler = DartLibraryCalls::LookupHandler(map_[i].port);
         OS::Print("Handler = %s\n", msg_handler.ToCString());
       }
     }
   }
 }

 }  // namespace dart
	// Copyright (c) 2012, 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/port.h"

	#include "platform/utils.h"
	#include "vm/dart_api_impl.h"
	#include "vm/dart_entry.h"
	#include "vm/isolate.h"
	#include "vm/lockers.h"
	#include "vm/message_handler.h"
	#include "vm/os_thread.h"

	namespace dart {

	Mutex* PortMap::mutex_ = NULL;
	PortMap::Entry* PortMap::map_ = NULL;
	MessageHandler* PortMap::deleted_entry_ = reinterpret_cast<MessageHandler*>(1);
	intptr_t PortMap::capacity_ = 0;
	intptr_t PortMap::used_ = 0;
	intptr_t PortMap::deleted_ = 0;
	Random* PortMap::prng_ = NULL;

	intptr_t PortMap::FindPort(Dart_Port port) {
	// ILLEGAL_PORT (0) is used as a sentinel value in Entry.port. The loop below
	// could return the index to a deleted port when we are searching for
	// port id ILLEGAL_PORT. Return -1 immediately to indicate the port
	// does not exist.
	if (port == ILLEGAL_PORT) {
	return -1;
	}
	ASSERT(port != ILLEGAL_PORT);
	intptr_t index = port % capacity_;
	intptr_t start_index = index;
	Entry entry = map_[index];
	while (entry.handler != NULL) {
	if (entry.port == port) {
	return index;
	}
	index = (index + 1) % capacity_;
	// Prevent endless loops.
	ASSERT(index != start_index);
	entry = map_[index];
	}
	return -1;
	}

	void PortMap::Rehash(intptr_t new_capacity) {
	Entry* new_ports = new Entry[new_capacity];
	memset(new_ports, 0, new_capacity * sizeof(Entry));

	for (intptr_t i = 0; i < capacity_; i++) {
	Entry entry = map_[i];
	// Skip free and deleted entries.
	if (entry.port != 0) {
	intptr_t new_index = entry.port % new_capacity;
	while (new_ports[new_index].port != 0) {
	new_index = (new_index + 1) % new_capacity;
	}
	new_ports[new_index] = entry;
	}
	}
	delete[] map_;
	map_ = new_ports;
	capacity_ = new_capacity;
	deleted_ = 0;
	}

	const char* PortMap::PortStateString(PortState kind) {
	switch (kind) {
	case kNewPort:
	return "new";
	case kLivePort:
	return "live";
	case kControlPort:
	return "control";
	default:
	UNREACHABLE();
	return "UNKNOWN";
	}
	}

	Dart_Port PortMap::AllocatePort() {
	const Dart_Port kMASK = 0x3fffffff;
	Dart_Port result = prng_->NextUInt32() & kMASK;

	// Keep getting new values while we have an illegal port number or the port
	// number is already in use.
	while ((result == 0) \|\| (FindPort(result) >= 0)) {
	result = prng_->NextUInt32() & kMASK;
	}

	ASSERT(result != 0);
	ASSERT(FindPort(result) < 0);
	return result;
	}

	void PortMap::SetPortState(Dart_Port port, PortState state) {
	MutexLocker ml(mutex_);
	intptr_t index = FindPort(port);
	ASSERT(index >= 0);
	PortState old_state = map_[index].state;
	ASSERT(old_state == kNewPort);
	map_[index].state = state;
	if (state == kLivePort) {
	map_[index].handler->increment_live_ports();
	}
	if (FLAG_trace_isolates) {
	OS::Print(
	"[^] Port (%s) -> (%s): \n"
	"\thandler: %s\n"
	"\tport: %" Pd64 "\n",
	PortStateString(old_state), PortStateString(state),
	map_[index].handler->name(), port);
	}
	}

	void PortMap::MaintainInvariants() {
	intptr_t empty = capacity_ - used_ - deleted_;
	if (used_ > ((capacity_ / 4) * 3)) {
	// Grow the port map.
	Rehash(capacity_ * 2);
	} else if (empty < deleted_) {
	// Rehash without growing the table to flush the deleted slots out of the
	// map.
	Rehash(capacity_);
	}
	}

	Dart_Port PortMap::CreatePort(MessageHandler* handler) {
	ASSERT(handler != NULL);
	MutexLocker ml(mutex_);
	#if defined(DEBUG)
	handler->CheckAccess();
	#endif

	Entry entry;
	entry.port = AllocatePort();
	entry.handler = handler;
	entry.state = kNewPort;

	// Search for the first unused slot. Make use of the knowledge that here is
	// currently no port with this id in the port map.
	ASSERT(FindPort(entry.port) < 0);
	intptr_t index = entry.port % capacity_;
	Entry cur = map_[index];
	// Stop the search at the first found unused (free or deleted) slot.
	while (cur.port != 0) {
	index = (index + 1) % capacity_;
	cur = map_[index];
	}

	// Insert the newly created port at the index.
	ASSERT(index >= 0);
	ASSERT(index < capacity_);
	ASSERT(map_[index].port == 0);
	ASSERT((map_[index].handler == NULL) \|\|
	(map_[index].handler == deleted_entry_));
	if (map_[index].handler == deleted_entry_) {
	// Consuming a deleted entry.
	deleted_--;
	}
	map_[index] = entry;

	// Increment number of used slots and grow if necessary.
	used_++;
	MaintainInvariants();

	if (FLAG_trace_isolates) {
	OS::Print(
	"[+] Opening port: \n"
	"\thandler: %s\n"
	"\tport: %" Pd64 "\n",
	handler->name(), entry.port);
	}

	return entry.port;
	}

	bool PortMap::ClosePort(Dart_Port port) {
	MessageHandler* handler = NULL;
	{
	MutexLocker ml(mutex_);
	intptr_t index = FindPort(port);
	if (index < 0) {
	return false;
	}
	ASSERT(index < capacity_);
	ASSERT(map_[index].port != 0);
	ASSERT(map_[index].handler != deleted_entry_);
	ASSERT(map_[index].handler != NULL);

	handler = map_[index].handler;
	#if defined(DEBUG)
	handler->CheckAccess();
	#endif
	// Before releasing the lock mark the slot in the map as deleted. This makes
	// it possible to release the port map lock before flushing all of its
	// pending messages below.
	map_[index].port = 0;
	map_[index].handler = deleted_entry_;
	if (map_[index].state == kLivePort) {
	handler->decrement_live_ports();
	}

	used_--;
	deleted_++;
	MaintainInvariants();
	}
	handler->ClosePort(port);
	if (!handler->HasLivePorts() && handler->OwnedByPortMap()) {
	// Delete handler as soon as it isn't busy with a task.
	handler->RequestDeletion();
	}
	return true;
	}

	void PortMap::ClosePorts(MessageHandler* handler) {
	{
	MutexLocker ml(mutex_);
	for (intptr_t i = 0; i < capacity_; i++) {
	if (map_[i].handler == handler) {
	// Mark the slot as deleted.
	map_[i].port = 0;
	map_[i].handler = deleted_entry_;
	if (map_[i].state == kLivePort) {
	handler->decrement_live_ports();
	}
	used_--;
	deleted_++;
	}
	}
	MaintainInvariants();
	}
	handler->CloseAllPorts();
	}

	bool PortMap::PostMessage(Message* message) {
	MutexLocker ml(mutex_);
	intptr_t index = FindPort(message->dest_port());
	if (index < 0) {
	delete message;
	return false;
	}
	ASSERT(index >= 0);
	ASSERT(index < capacity_);
	MessageHandler* handler = map_[index].handler;
	ASSERT(map_[index].port != 0);
	ASSERT((handler != NULL) && (handler != deleted_entry_));
	handler->PostMessage(message);
	return true;
	}

	bool PortMap::IsLocalPort(Dart_Port id) {
	MutexLocker ml(mutex_);
	intptr_t index = FindPort(id);
	if (index < 0) {
	// Port does not exist.
	return false;
	}

	MessageHandler* handler = map_[index].handler;
	return handler->IsCurrentIsolate();
	}

	Isolate* PortMap::GetIsolate(Dart_Port id) {
	MutexLocker ml(mutex_);
	intptr_t index = FindPort(id);
	if (index < 0) {
	// Port does not exist.
	return NULL;
	}

	MessageHandler* handler = map_[index].handler;
	return handler->isolate();
	}

	void PortMap::InitOnce() {
	mutex_ = new Mutex();
	prng_ = new Random();

	static const intptr_t kInitialCapacity = 8;
	// TODO(iposva): Verify whether we want to keep exponentially growing.
	ASSERT(Utils::IsPowerOfTwo(kInitialCapacity));
	map_ = new Entry[kInitialCapacity];
	memset(map_, 0, kInitialCapacity * sizeof(Entry));
	capacity_ = kInitialCapacity;
	used_ = 0;
	deleted_ = 0;
	}

	void PortMap::PrintPortsForMessageHandler(MessageHandler* handler,
	JSONStream* stream) {
	#ifndef PRODUCT
	if (!FLAG_support_service) {
	return;
	}
	JSONObject jsobj(stream);
	jsobj.AddProperty("type", "_Ports");
	Object& msg_handler = Object::Handle();
	{
	JSONArray ports(&jsobj, "ports");
	SafepointMutexLocker ml(mutex_);
	for (intptr_t i = 0; i < capacity_; i++) {
	if (map_[i].handler == handler) {
	if (map_[i].state == kLivePort) {
	JSONObject port(&ports);
	port.AddProperty("type", "_Port");
	port.AddPropertyF("name", "Isolate Port (%" Pd64 ")", map_[i].port);
	msg_handler = DartLibraryCalls::LookupHandler(map_[i].port);
	port.AddProperty("handler", msg_handler);
	}
	}
	}
	}
	#endif
	}

	void PortMap::DebugDumpForMessageHandler(MessageHandler* handler) {
	SafepointMutexLocker ml(mutex_);
	Object& msg_handler = Object::Handle();
	for (intptr_t i = 0; i < capacity_; i++) {
	if (map_[i].handler == handler) {
	if (map_[i].state == kLivePort) {
	OS::Print("Live Port = %" Pd64 "\n", map_[i].port);
	msg_handler = DartLibraryCalls::LookupHandler(map_[i].port);
	OS::Print("Handler = %s\n", msg_handler.ToCString());
	}
	}
	}
	}

	} // namespace dart