tools/android/forwarder/forwarder.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include <errno.h>
   6 #include <fcntl.h>
   7 #include <netinet/in.h>
   8 #include <netinet/tcp.h>
   9 #include <pthread.h>
  10 #include <signal.h>
  11 #include <stdio.h>
  12 #include <stdlib.h>
  13 #include <string.h>
  14 #include <sys/select.h>
  15 #include <sys/socket.h>
  16 #include <sys/wait.h>
  17 #include <unistd.h>
  18
  19 #include "base/command_line.h"
  20 #include "base/logging.h"
  21 #include "base/posix/eintr_wrapper.h"
  22 #include "tools/android/common/adb_connection.h"
  23 #include "tools/android/common/daemon.h"
  24 #include "tools/android/common/net.h"
  25
  26 namespace {
  27
  28 const pthread_t kInvalidThread = static_cast<pthread_t>(-1);
  29 volatile bool g_killed = false;
  30
  31 void CloseSocket(int fd) {
  32   if (fd >= 0) {
  33     int old_errno = errno;
  34     (void) HANDLE_EINTR(close(fd));
  35     errno = old_errno;
  36   }
  37 }
  38
  39 class Buffer {
  40  public:
  41   Buffer()
  42       : bytes_read_(0),
  43         write_offset_(0) {
  44   }
  45
  46   bool CanRead() {
  47     return bytes_read_ == 0;
  48   }
  49
  50   bool CanWrite() {
  51     return write_offset_ < bytes_read_;
  52   }
  53
  54   int Read(int fd) {
  55     int ret = -1;
  56     if (CanRead()) {
  57       ret = HANDLE_EINTR(read(fd, buffer_, kBufferSize));
  58       if (ret > 0)
  59         bytes_read_ = ret;
  60     }
  61     return ret;
  62   }
  63
  64   int Write(int fd) {
  65     int ret = -1;
  66     if (CanWrite()) {
  67       ret = HANDLE_EINTR(write(fd, buffer_ + write_offset_,
  68                                bytes_read_ - write_offset_));
  69       if (ret > 0) {
  70         write_offset_ += ret;
  71         if (write_offset_ == bytes_read_) {
  72           write_offset_ = 0;
  73           bytes_read_ = 0;
  74         }
  75       }
  76     }
  77     return ret;
  78   }
  79
  80  private:
  81   // A big buffer to let our file-over-http bridge work more like real file.
  82   static const int kBufferSize = 1024 * 128;
  83   int bytes_read_;
  84   int write_offset_;
  85   char buffer_[kBufferSize];
  86
  87   DISALLOW_COPY_AND_ASSIGN(Buffer);
  88 };
  89
  90 class Server;
  91
  92 struct ForwarderThreadInfo {
  93   ForwarderThreadInfo(Server* a_server, int a_forwarder_index)
  94       : server(a_server),
  95         forwarder_index(a_forwarder_index) {
  96   }
  97   Server* server;
  98   int forwarder_index;
  99 };
 100
 101 struct ForwarderInfo {
 102   time_t start_time;
 103   int socket1;
 104   time_t socket1_last_byte_time;
 105   size_t socket1_bytes;
 106   int socket2;
 107   time_t socket2_last_byte_time;
 108   size_t socket2_bytes;
 109 };
 110
 111 class Server {
 112  public:
 113   Server()
 114       : thread_(kInvalidThread),
 115         socket_(-1) {
 116     memset(forward_to_, 0, sizeof(forward_to_));
 117     memset(&forwarders_, 0, sizeof(forwarders_));
 118   }
 119
 120   int GetFreeForwarderIndex() {
 121     for (int i = 0; i < kMaxForwarders; i++) {
 122       if (forwarders_[i].start_time == 0)
 123         return i;
 124     }
 125     return -1;
 126   }
 127
 128   void DisposeForwarderInfo(int index) {
 129     forwarders_[index].start_time = 0;
 130   }
 131
 132   ForwarderInfo* GetForwarderInfo(int index) {
 133     return &forwarders_[index];
 134   }
 135
 136   void DumpInformation() {
 137     LOG(INFO) << "Server information: " << forward_to_;
 138     LOG(INFO) << "No.: age up(bytes,idle) down(bytes,idle)";
 139     int count = 0;
 140     time_t now = time(NULL);
 141     for (int i = 0; i < kMaxForwarders; i++) {
 142       const ForwarderInfo& info = forwarders_[i];
 143       if (info.start_time) {
 144         count++;
 145         LOG(INFO) << count << ": " << now - info.start_time << " up("
 146                   << info.socket1_bytes << ","
 147                   << now - info.socket1_last_byte_time << " down("
 148                   << info.socket2_bytes << ","
 149                   << now - info.socket2_last_byte_time << ")";
 150       }
 151     }
 152   }
 153
 154   void Shutdown() {
 155     if (socket_ >= 0)
 156       shutdown(socket_, SHUT_RDWR);
 157   }
 158
 159   bool InitSocket(const char* arg);
 160
 161   void StartThread() {
 162     pthread_create(&thread_, NULL, ServerThread, this);
 163   }
 164
 165   void JoinThread() {
 166     if (thread_ != kInvalidThread)
 167       pthread_join(thread_, NULL);
 168   }
 169
 170  private:
 171   static void* ServerThread(void* arg);
 172
 173   // There are 3 kinds of threads that will access the array:
 174   // 1. Server thread will get a free ForwarderInfo and initialize it;
 175   // 2. Forwarder threads will dispose the ForwarderInfo when it finishes;
 176   // 3. Main thread will iterate and print the forwarders.
 177   // Using an array is not optimal, but can avoid locks or other complex
 178   // inter-thread communication.
 179   static const int kMaxForwarders = 512;
 180   ForwarderInfo forwarders_[kMaxForwarders];
 181
 182   pthread_t thread_;
 183   int socket_;
 184   char forward_to_[40];
 185
 186   DISALLOW_COPY_AND_ASSIGN(Server);
 187 };
 188
 189 // Forwards all outputs from one socket to another socket.
 190 void* ForwarderThread(void* arg) {
 191   ForwarderThreadInfo* thread_info =
 192       reinterpret_cast<ForwarderThreadInfo*>(arg);
 193   Server* server = thread_info->server;
 194   int index = thread_info->forwarder_index;
 195   delete thread_info;
 196   ForwarderInfo* info = server->GetForwarderInfo(index);
 197   int socket1 = info->socket1;
 198   int socket2 = info->socket2;
 199   int nfds = socket1 > socket2 ? socket1 + 1 : socket2 + 1;
 200   fd_set read_fds;
 201   fd_set write_fds;
 202   Buffer buffer1;
 203   Buffer buffer2;
 204
 205   while (!g_killed) {
 206     FD_ZERO(&read_fds);
 207     if (buffer1.CanRead())
 208       FD_SET(socket1, &read_fds);
 209     if (buffer2.CanRead())
 210       FD_SET(socket2, &read_fds);
 211
 212     FD_ZERO(&write_fds);
 213     if (buffer1.CanWrite())
 214       FD_SET(socket2, &write_fds);
 215     if (buffer2.CanWrite())
 216       FD_SET(socket1, &write_fds);
 217
 218     if (HANDLE_EINTR(select(nfds, &read_fds, &write_fds, NULL, NULL)) <= 0) {
 219       LOG(ERROR) << "Select error: " << strerror(errno);
 220       break;
 221     }
 222
 223     int now = time(NULL);
 224     if (FD_ISSET(socket1, &read_fds)) {
 225       info->socket1_last_byte_time = now;
 226       int bytes = buffer1.Read(socket1);
 227       if (bytes <= 0)
 228         break;
 229       info->socket1_bytes += bytes;
 230     }
 231     if (FD_ISSET(socket2, &read_fds)) {
 232       info->socket2_last_byte_time = now;
 233       int bytes = buffer2.Read(socket2);
 234       if (bytes <= 0)
 235         break;
 236       info->socket2_bytes += bytes;
 237     }
 238     if (FD_ISSET(socket1, &write_fds)) {
 239       if (buffer2.Write(socket1) <= 0)
 240         break;
 241     }
 242     if (FD_ISSET(socket2, &write_fds)) {
 243       if (buffer1.Write(socket2) <= 0)
 244         break;
 245     }
 246   }
 247
 248   CloseSocket(socket1);
 249   CloseSocket(socket2);
 250   server->DisposeForwarderInfo(index);
 251   return NULL;
 252 }
 253
 254 // Listens to a server socket. On incoming request, forward it to the host.
 255 // static
 256 void* Server::ServerThread(void* arg) {
 257   Server* server = reinterpret_cast<Server*>(arg);
 258   while (!g_killed) {
 259     int forwarder_index = server->GetFreeForwarderIndex();
 260     if (forwarder_index < 0) {
 261       LOG(ERROR) << "Too many forwarders";
 262       continue;
 263     }
 264
 265     struct sockaddr_in addr;
 266     socklen_t addr_len = sizeof(addr);
 267     int socket = HANDLE_EINTR(accept(server->socket_,
 268                                      reinterpret_cast<sockaddr*>(&addr),
 269                                      &addr_len));
 270     if (socket < 0) {
 271       LOG(ERROR) << "Failed to accept: " << strerror(errno);
 272       break;
 273     }
 274     tools::DisableNagle(socket);
 275
 276     int host_socket = tools::ConnectAdbHostSocket(server->forward_to_);
 277     if (host_socket >= 0) {
 278       // Set NONBLOCK flag because we use select().
 279       fcntl(socket, F_SETFL, fcntl(socket, F_GETFL) | O_NONBLOCK);
 280       fcntl(host_socket, F_SETFL, fcntl(host_socket, F_GETFL) | O_NONBLOCK);
 281
 282       ForwarderInfo* forwarder_info = server->GetForwarderInfo(forwarder_index);
 283       time_t now = time(NULL);
 284       forwarder_info->start_time = now;
 285       forwarder_info->socket1 = socket;
 286       forwarder_info->socket1_last_byte_time = now;
 287       forwarder_info->socket1_bytes = 0;
 288       forwarder_info->socket2 = host_socket;
 289       forwarder_info->socket2_last_byte_time = now;
 290       forwarder_info->socket2_bytes = 0;
 291
 292       pthread_t thread;
 293       pthread_create(&thread, NULL, ForwarderThread,
 294                      new ForwarderThreadInfo(server, forwarder_index));
 295     } else {
 296       // Close the unused client socket which is failed to connect to host.
 297       CloseSocket(socket);
 298     }
 299   }
 300
 301   CloseSocket(server->socket_);
 302   server->socket_ = -1;
 303   return NULL;
 304 }
 305
 306 // Format of arg: <Device port>[:<Forward to port>:<Forward to address>]
 307 bool Server::InitSocket(const char* arg) {
 308   char* endptr;
 309   int local_port = static_cast<int>(strtol(arg, &endptr, 10));
 310   if (local_port < 0)
 311     return false;
 312
 313   if (*endptr != ':') {
 314     snprintf(forward_to_, sizeof(forward_to_), "%d:127.0.0.1", local_port);
 315   } else {
 316     strncpy(forward_to_, endptr + 1, sizeof(forward_to_) - 1);
 317   }
 318
 319   socket_ = socket(AF_INET, SOCK_STREAM, 0);
 320   if (socket_ < 0) {
 321     perror("server socket");
 322     return false;
 323   }
 324   tools::DisableNagle(socket_);
 325
 326   sockaddr_in addr;
 327   memset(&addr, 0, sizeof(addr));
 328   addr.sin_family = AF_INET;
 329   addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
 330   addr.sin_port = htons(local_port);
 331   int reuse_addr = 1;
 332   setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR,
 333              &reuse_addr, sizeof(reuse_addr));
 334   tools::DeferAccept(socket_);
 335   if (HANDLE_EINTR(bind(socket_, reinterpret_cast<sockaddr*>(&addr),
 336                         sizeof(addr))) < 0 ||
 337       HANDLE_EINTR(listen(socket_, 5)) < 0) {
 338     perror("server bind");
 339     CloseSocket(socket_);
 340     socket_ = -1;
 341     return false;
 342   }
 343
 344   if (local_port == 0) {
 345     socklen_t addrlen = sizeof(addr);
 346     if (getsockname(socket_, reinterpret_cast<sockaddr*>(&addr), &addrlen)
 347         != 0) {
 348       perror("get listen address");
 349       CloseSocket(socket_);
 350       socket_ = -1;
 351       return false;
 352     }
 353     local_port = ntohs(addr.sin_port);
 354   }
 355
 356   printf("Forwarding device port %d to host %s\n", local_port, forward_to_);
 357   return true;
 358 }
 359
 360 int g_server_count = 0;
 361 Server* g_servers = NULL;
 362
 363 void KillHandler(int unused) {
 364   g_killed = true;
 365   for (int i = 0; i < g_server_count; i++)
 366     g_servers[i].Shutdown();
 367 }
 368
 369 void DumpInformation(int unused) {
 370   for (int i = 0; i < g_server_count; i++)
 371     g_servers[i].DumpInformation();
 372 }
 373
 374 }  // namespace
 375
 376 int main(int argc, char** argv) {
 377   printf("Android device to host TCP forwarder\n");
 378   printf("Like 'adb forward' but in the reverse direction\n");
 379
 380   CommandLine command_line(argc, argv);
 381   CommandLine::StringVector server_args = command_line.GetArgs();
 382   if (tools::HasHelpSwitch(command_line) || server_args.empty()) {
 383     tools::ShowHelp(
 384         argv[0],
 385         "<Device port>[:<Forward to port>:<Forward to address>] ...",
 386         "  <Forward to port> default is <Device port>\n"
 387         "  <Forward to address> default is 127.0.0.1\n"
 388         "If <Device port> is 0, a port will by dynamically allocated.\n");
 389     return 0;
 390   }
 391
 392   g_servers = new Server[server_args.size()];
 393   g_server_count = 0;
 394   int failed_count = 0;
 395   for (size_t i = 0; i < server_args.size(); i++) {
 396     if (!g_servers[g_server_count].InitSocket(server_args[i].c_str())) {
 397       printf("Couldn't start forwarder server for port spec: %s\n",
 398              server_args[i].c_str());
 399       ++failed_count;
 400     } else {
 401       ++g_server_count;
 402     }
 403   }
 404
 405   if (g_server_count == 0) {
 406     printf("No forwarder servers could be started. Exiting.\n");
 407     delete [] g_servers;
 408     return failed_count;
 409   }
 410
 411   if (!tools::HasNoSpawnDaemonSwitch(command_line))
 412     tools::SpawnDaemon(failed_count);
 413
 414   signal(SIGTERM, KillHandler);
 415   signal(SIGUSR2, DumpInformation);
 416
 417   for (int i = 0; i < g_server_count; i++)
 418     g_servers[i].StartThread();
 419   for (int i = 0; i < g_server_count; i++)
 420     g_servers[i].JoinThread();
 421   g_server_count = 0;
 422   delete [] g_servers;
 423
 424   return 0;
 425 }
 426