Use hash map for event handler file descriptor map

runtime/bin/eventhandler_linux.cc

 // 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 <errno.h>
 #include <poll.h>
 #include <pthread.h>
 #include <stdio.h>
 #include <string.h>
 #include <sys/time.h>
 #include <unistd.h>
 
 #include "bin/eventhandler.h"
 #include "bin/fdutils.h"
+#include "bin/hashmap.h"
+#include "platform/utils.h"
 
 
 int64_t GetCurrentTimeMilliseconds() {
   struct timeval tv;
   if (gettimeofday(&tv, NULL) < 0) {
     UNREACHABLE();
     return 0;
   }
   return ((static_cast<int64_t>(tv.tv_sec) * 1000000) + tv.tv_usec) / 1000;
 }
 
 
-static const int kInitialPortMapSize = 16;
-static const int kPortMapGrowingFactor = 2;
 static const int kInterruptMessageSize = sizeof(InterruptMessage);
 static const int kInfinityTimeout = -1;
 static const int kTimerId = -1;
 
 
 intptr_t SocketData::GetPollEvents() {
   // Do not ask for POLLERR and POLLHUP explicitly as they are
   // triggered anyway.
   intptr_t events = 0;
   if (!IsClosedRead()) {
     if ((mask_ & (1 << kInEvent)) != 0) {
       events |= POLLIN;
     }
   }
   if (!IsClosedWrite()) {
     if ((mask_ & (1 << kOutEvent)) != 0) {
       events |= POLLOUT;
     }
   }
   return events;
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
   intptr_t result;
-  socket_map_size_ = kInitialPortMapSize;
-  socket_map_ = reinterpret_cast<SocketData*>(calloc(socket_map_size_,
-                                                     sizeof(SocketData)));
-  ASSERT(socket_map_ != NULL);
   result = TEMP_FAILURE_RETRY(pipe(interrupt_fds_));
   if (result != 0) {
     FATAL("Pipe creation failed");
   }
   FDUtils::SetNonBlocking(interrupt_fds_[0]);
   timeout_ = kInfinityTimeout;
   timeout_port_ = 0;
 }
 
 
 EventHandlerImplementation::~EventHandlerImplementation() {
-  free(socket_map_);
   TEMP_FAILURE_RETRY(close(interrupt_fds_[0]));
   TEMP_FAILURE_RETRY(close(interrupt_fds_[1]));
 }
 
 
-// TODO(hpayer): Use hash table instead of array.
 SocketData* EventHandlerImplementation::GetSocketData(intptr_t fd) {
   ASSERT(fd >= 0);
-  if (fd >= socket_map_size_) {
-    intptr_t new_socket_map_size = socket_map_size_;
-    do {
-      new_socket_map_size = new_socket_map_size * kPortMapGrowingFactor;
-    } while (fd >= new_socket_map_size);
-    size_t new_socket_map_bytes = new_socket_map_size * sizeof(SocketData);
-    socket_map_ = reinterpret_cast<SocketData*>(realloc(socket_map_,
-                                                        new_socket_map_bytes));
-    ASSERT(socket_map_ != NULL);
-    size_t socket_map_bytes = socket_map_size_ * sizeof(SocketData);
-    memset(socket_map_ + socket_map_size_,
-           0,
-           new_socket_map_bytes - socket_map_bytes);
-    socket_map_size_ = new_socket_map_size;
+  HashMap::Entry* entry = socket_map_.Lookup(
+      GetHashmapKeyFromFd(fd), GetHashmapHashFromFd(fd), true);
+  ASSERT(entry != NULL);
+  SocketData* sd = reinterpret_cast<SocketData*>(entry->value);
+  if (sd == NULL) {
+    sd = new SocketData(fd);

[Ivan Posva, 2012/01/20 21:42:57]

Please add a comment making it clear that you are now populating the entry in
the socket_map_.

[Søren Gjesse, 2012/01/23 09:03:49]

Done.

+    entry->value = sd;
   }
-
-  SocketData* sd = socket_map_ + fd;
-  sd->set_fd(fd);  // For now just make sure the fd is set.
+  ASSERT(fd == sd->fd());
   return sd;
 }
 
 
 void EventHandlerImplementation::WakeupHandler(intptr_t id,
                                                Dart_Port dart_port,
                                                int64_t data) {
   InterruptMessage msg;
   msg.id = id;
   msg.dart_port = dart_port;
   msg.data = data;
   intptr_t result =
       FDUtils::WriteToBlocking(interrupt_fds_[1], &msg, kInterruptMessageSize);
   if (result != kInterruptMessageSize) {
     FATAL("Interrupt message failure");
   }
 }
 
 
 struct pollfd* EventHandlerImplementation::GetPollFds(intptr_t* pollfds_size) {
   struct pollfd* pollfds;
 
   // Calculate the number of file descriptors to poll on.
   intptr_t numPollfds = 1;
-  for (int i = 0; i < socket_map_size_; i++) {
-    SocketData* sd = &socket_map_[i];
+  for (HashMap::Entry* entry = socket_map_.Start();
+       entry != NULL;
+       entry = socket_map_.Next(entry)) {
+    SocketData* sd = reinterpret_cast<SocketData*>(entry->value);
     if (sd->port() > 0 && sd->GetPollEvents() != 0) numPollfds++;
   }
 
   pollfds = reinterpret_cast<struct pollfd*>(calloc(sizeof(struct pollfd),
                                                     numPollfds));
   pollfds[0].fd = interrupt_fds_[0];
   pollfds[0].events |= POLLIN;
 
-  // TODO(hpayer): optimize the following iteration over the hash map
-  int j = 1;
-  for (int i = 0; i < socket_map_size_; i++) {
-    SocketData* sd = &socket_map_[i];
+  int i = 1;
+  for (HashMap::Entry* entry = socket_map_.Start();
+       entry != NULL;
+       entry = socket_map_.Next(entry)) {
+    SocketData* sd = reinterpret_cast<SocketData*>(entry->value);
     intptr_t events = sd->GetPollEvents();
     if (sd->port() > 0 && events != 0) {
       // Fd is added to the poll set.
-      pollfds[j].fd = sd->fd();
-      pollfds[j].events = events;
-      j++;
+      pollfds[i].fd = sd->fd();
+      pollfds[i].events = events;
+      i++;
     }
   }
-  ASSERT(numPollfds == j);
-  *pollfds_size = j;
+  ASSERT(numPollfds == i);
+  *pollfds_size = i;
+
   return pollfds;
 }
 
 
 bool EventHandlerImplementation::GetInterruptMessage(InterruptMessage* msg) {
   int total_read = 0;
   int bytes_read =
       TEMP_FAILURE_RETRY(read(interrupt_fds_[0], msg, kInterruptMessageSize));
   if (bytes_read < 0) {
     return false;
@@ skipping 22 matching lines @@
         ASSERT(msg.data == (1 << kShutdownReadCommand));
         // Close the socket for reading.
         sd->ShutdownRead();
       } else if ((msg.data & (1 << kShutdownWriteCommand)) != 0) {
         ASSERT(msg.data == (1 << kShutdownWriteCommand));
         // Close the socket for writing.
         sd->ShutdownWrite();
       } else if ((msg.data & (1 << kCloseCommand)) != 0) {
         ASSERT(msg.data == (1 << kCloseCommand));
         // Close the socket and free system resources.
+        intptr_t fd = sd->fd();
         sd->Close();
+        socket_map_.Remove(GetHashmapKeyFromFd(fd), GetHashmapHashFromFd(fd));
+        delete sd;
       } else {
         // Setup events to wait for.
         sd->SetPortAndMask(msg.dart_port, msg.data);
       }
     }
   }
 }
 
 #ifdef DEBUG_POLL
 static void PrintEventMask(struct pollfd* pollfd) {
@@ skipping 178 matching lines @@
     FATAL("Create start event handler thread");
   }
 }
 
 
 void EventHandlerImplementation::SendData(intptr_t id,
                                           Dart_Port dart_port,
                                           intptr_t data) {
   WakeupHandler(id, dart_port, data);
 }
+
+
+void* EventHandlerImplementation::GetHashmapKeyFromFd(intptr_t fd) {
+  // The hashmap does not support keys with value 0.
+  return reinterpret_cast<void*>(fd + 1);
+}
+
+
+uint32_t EventHandlerImplementation::GetHashmapHashFromFd(intptr_t fd) {
+  // The hashmap does not support keys with value 0.
+  return dart::Utils::WordHash(fd + 1);
+}