Home | History | Annotate | Download | only in adb 
      1 /*
      2  * Copyright (C) 2015 The Android Open Source Project
      3  *
      4  * Licensed under the Apache License, Version 2.0 (the "License");
      5  * you may not use this file except in compliance with the License.
      6  * You may obtain a copy of the License at
      7  *
      8  *      http://www.apache.org/licenses/LICENSE-2.0
      9  *
     10  * Unless required by applicable law or agreed to in writing, software
     11  * distributed under the License is distributed on an "AS IS" BASIS,
     12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     13  * See the License for the specific language governing permissions and
     14  * limitations under the License.
     15  */
     16 
     17 #include "fdevent.h"
     18 
     19 #include <gtest/gtest.h>
     20 
     21 #include <limits>
     22 #include <queue>
     23 #include <string>
     24 #include <thread>
     25 #include <vector>
     26 
     27 #include "adb_io.h"
     28 #include "fdevent_test.h"
     29 #include "sysdeps/memory.h"
     30 
     31 class FdHandler {
     32   public:
     33     FdHandler(int read_fd, int write_fd) : read_fd_(read_fd), write_fd_(write_fd) {
     34         fdevent_install(&read_fde_, read_fd_, FdEventCallback, this);
     35         fdevent_add(&read_fde_, FDE_READ);
     36         fdevent_install(&write_fde_, write_fd_, FdEventCallback, this);
     37     }
     38 
     39     ~FdHandler() {
     40         fdevent_remove(&read_fde_);
     41         fdevent_remove(&write_fde_);
     42     }
     43 
     44   private:
     45     static void FdEventCallback(int fd, unsigned events, void* userdata) {
     46         FdHandler* handler = reinterpret_cast<FdHandler*>(userdata);
     47         ASSERT_EQ(0u, (events & ~(FDE_READ | FDE_WRITE))) << "unexpected events: " << events;
     48         if (events & FDE_READ) {
     49             ASSERT_EQ(fd, handler->read_fd_);
     50             char c;
     51             ASSERT_EQ(1, adb_read(fd, &c, 1));
     52             handler->queue_.push(c);
     53             fdevent_add(&handler->write_fde_, FDE_WRITE);
     54         }
     55         if (events & FDE_WRITE) {
     56             ASSERT_EQ(fd, handler->write_fd_);
     57             ASSERT_FALSE(handler->queue_.empty());
     58             char c = handler->queue_.front();
     59             handler->queue_.pop();
     60             ASSERT_EQ(1, adb_write(fd, &c, 1));
     61             if (handler->queue_.empty()) {
     62               fdevent_del(&handler->write_fde_, FDE_WRITE);
     63             }
     64         }
     65     }
     66 
     67   private:
     68     const int read_fd_;
     69     const int write_fd_;
     70     fdevent read_fde_;
     71     fdevent write_fde_;
     72     std::queue<char> queue_;
     73 };
     74 
     75 struct ThreadArg {
     76     int first_read_fd;
     77     int last_write_fd;
     78     size_t middle_pipe_count;
     79 };
     80 
     81 TEST_F(FdeventTest, fdevent_terminate) {
     82     PrepareThread();
     83 
     84     std::thread thread(fdevent_loop);
     85     TerminateThread(thread);
     86 }
     87 
     88 static void FdEventThreadFunc(ThreadArg* arg) {
     89     std::vector<int> read_fds;
     90     std::vector<int> write_fds;
     91 
     92     read_fds.push_back(arg->first_read_fd);
     93     for (size_t i = 0; i < arg->middle_pipe_count; ++i) {
     94         int fds[2];
     95         ASSERT_EQ(0, adb_socketpair(fds));
     96         read_fds.push_back(fds[0]);
     97         write_fds.push_back(fds[1]);
     98     }
     99     write_fds.push_back(arg->last_write_fd);
    100 
    101     std::vector<std::unique_ptr<FdHandler>> fd_handlers;
    102     for (size_t i = 0; i < read_fds.size(); ++i) {
    103         fd_handlers.push_back(std::make_unique<FdHandler>(read_fds[i], write_fds[i]));
    104     }
    105 
    106     fdevent_loop();
    107 }
    108 
    109 TEST_F(FdeventTest, smoke) {
    110     const size_t PIPE_COUNT = 10;
    111     const size_t MESSAGE_LOOP_COUNT = 100;
    112     const std::string MESSAGE = "fdevent_test";
    113     int fd_pair1[2];
    114     int fd_pair2[2];
    115     ASSERT_EQ(0, adb_socketpair(fd_pair1));
    116     ASSERT_EQ(0, adb_socketpair(fd_pair2));
    117     ThreadArg thread_arg;
    118     thread_arg.first_read_fd = fd_pair1[0];
    119     thread_arg.last_write_fd = fd_pair2[1];
    120     thread_arg.middle_pipe_count = PIPE_COUNT;
    121     int writer = fd_pair1[1];
    122     int reader = fd_pair2[0];
    123 
    124     PrepareThread();
    125     std::thread thread(FdEventThreadFunc, &thread_arg);
    126 
    127     for (size_t i = 0; i < MESSAGE_LOOP_COUNT; ++i) {
    128         std::string read_buffer = MESSAGE;
    129         std::string write_buffer(MESSAGE.size(), 'a');
    130         ASSERT_TRUE(WriteFdExactly(writer, read_buffer.c_str(), read_buffer.size()));
    131         ASSERT_TRUE(ReadFdExactly(reader, &write_buffer[0], write_buffer.size()));
    132         ASSERT_EQ(read_buffer, write_buffer);
    133     }
    134 
    135     TerminateThread(thread);
    136     ASSERT_EQ(0, adb_close(writer));
    137     ASSERT_EQ(0, adb_close(reader));
    138 }
    139 
    140 struct InvalidFdArg {
    141     fdevent fde;
    142     unsigned expected_events;
    143     size_t* happened_event_count;
    144 };
    145 
    146 static void InvalidFdEventCallback(int fd, unsigned events, void* userdata) {
    147     InvalidFdArg* arg = reinterpret_cast<InvalidFdArg*>(userdata);
    148     ASSERT_EQ(arg->expected_events, events);
    149     fdevent_remove(&arg->fde);
    150     if (++*(arg->happened_event_count) == 2) {
    151         fdevent_terminate_loop();
    152     }
    153 }
    154 
    155 static void InvalidFdThreadFunc() {
    156     const int INVALID_READ_FD = std::numeric_limits<int>::max() - 1;
    157     size_t happened_event_count = 0;
    158     InvalidFdArg read_arg;
    159     read_arg.expected_events = FDE_READ | FDE_ERROR;
    160     read_arg.happened_event_count = &happened_event_count;
    161     fdevent_install(&read_arg.fde, INVALID_READ_FD, InvalidFdEventCallback, &read_arg);
    162     fdevent_add(&read_arg.fde, FDE_READ);
    163 
    164     const int INVALID_WRITE_FD = std::numeric_limits<int>::max();
    165     InvalidFdArg write_arg;
    166     write_arg.expected_events = FDE_READ | FDE_ERROR;
    167     write_arg.happened_event_count = &happened_event_count;
    168     fdevent_install(&write_arg.fde, INVALID_WRITE_FD, InvalidFdEventCallback, &write_arg);
    169     fdevent_add(&write_arg.fde, FDE_WRITE);
    170     fdevent_loop();
    171 }
    172 
    173 TEST_F(FdeventTest, invalid_fd) {
    174     std::thread thread(InvalidFdThreadFunc);
    175     thread.join();
    176 }
    177 
    178 TEST_F(FdeventTest, run_on_main_thread) {
    179     std::vector<int> vec;
    180 
    181     PrepareThread();
    182     std::thread thread(fdevent_loop);
    183 
    184     // Block the main thread for a long time while we queue our callbacks.
    185     fdevent_run_on_main_thread([]() {
    186         check_main_thread();
    187         std::this_thread::sleep_for(std::chrono::seconds(1));
    188     });
    189 
    190     for (int i = 0; i < 1000000; ++i) {
    191         fdevent_run_on_main_thread([i, &vec]() {
    192             check_main_thread();
    193             vec.push_back(i);
    194         });
    195     }
    196 
    197     TerminateThread(thread);
    198 
    199     ASSERT_EQ(1000000u, vec.size());
    200     for (int i = 0; i < 1000000; ++i) {
    201         ASSERT_EQ(i, vec[i]);
    202     }
    203 }
    204 
    205 static std::function<void()> make_appender(std::vector<int>* vec, int value) {
    206     return [vec, value]() {
    207         check_main_thread();
    208         if (value == 100) {
    209             return;
    210         }
    211 
    212         vec->push_back(value);
    213         fdevent_run_on_main_thread(make_appender(vec, value + 1));
    214     };
    215 }
    216 
    217 TEST_F(FdeventTest, run_on_main_thread_reentrant) {
    218     std::vector<int> vec;
    219 
    220     PrepareThread();
    221     std::thread thread(fdevent_loop);
    222 
    223     fdevent_run_on_main_thread(make_appender(&vec, 0));
    224 
    225     TerminateThread(thread);
    226 
    227     ASSERT_EQ(100u, vec.size());
    228     for (int i = 0; i < 100; ++i) {
    229         ASSERT_EQ(i, vec[i]);
    230     }
    231 }
    232