You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tvm.apache.org by GitBox <gi...@apache.org> on 2021/06/15 22:11:44 UTC
[GitHub] [tvm] areusch commented on a change in pull request #7876: [iOS] Add tracker support into ios-rpc application
areusch commented on a change in pull request #7876:
URL: https://github.com/apache/tvm/pull/7876#discussion_r652185799
##########
File path: apps/ios_rpc/tvmrpc/RPCArgs.mm
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#import "RPCArgs.h"
+
+#import <Foundation/Foundation.h>
+
+#import "../../../src/support/socket.h"
+#import "../../../src/support/utils.h"
+
+#import <string>
+
+using std::string;
+
+const char* kUsage =
+ "\n"
+ "iOS tvmrpc application supported flags:\n"
+ "--host_url The tracker/proxy address, Default=0.0.0.0\n"
+ "--host_port The tracker/proxy port, Default=9190\n"
+ "--key The key used to identify the device type in tracker. Default=\"\"\n"
+ "--custom_addr Custom IP Address to Report to RPC Tracker. Default=\"\"\n"
+ "--immediate_connect No UI interconnection, connect to tracker immediately. Default=False\n"
+ "--verbose Allow to print status info to std out. Default=False\n"
+ "--server_mode Server mode. Can be \"pure_server\", \"proxy\" or \"tracker\". "
+ "Default=pure_server \n"
+ "\n";
+
+struct RPCArgs_cpp {
+ string host_url = "0.0.0.0";
+ int host_port = 9190;
+
+ string key;
+ string custom_addr = "";
+
+ bool immediate_connect = false;
+ bool verbose = false;
+ char server_mode = 0;
+
+ operator RPCArgs() const {
+ return RPCArgs{.host_url = host_url.c_str(),
+ .host_port = host_port,
+ .key = key.c_str(),
+ .custom_addr = custom_addr.c_str(),
+ .verbose = verbose,
+ .immediate_connect = immediate_connect,
+ .server_mode = server_mode};
+ };
+
+ RPCArgs_cpp& operator=(const RPCArgs& args) {
+ host_url = args.host_url;
+ host_port = args.host_port;
+ key = args.key;
+ custom_addr = args.custom_addr;
+ verbose = args.verbose;
+ immediate_connect = args.immediate_connect;
+ server_mode = args.server_mode;
+ return *this;
+ }
+};
+
+struct RPCArgs_cpp g_rpc_args;
+
+static void restore_from_cache() {
+ NSUserDefaults* defaults = [NSUserDefaults standardUserDefaults];
+
+ auto get_string_from_cache = [defaults](const char* key) {
+ NSString* ns_key = [NSString stringWithUTF8String:key];
+ NSString* ns_val = [defaults stringForKey:ns_key];
+ return std::string(ns_val != nil ? [ns_val UTF8String] : "");
+ };
+
+ auto get_int_from_cache = [defaults](const char* key) {
+ NSString* ns_key = [NSString stringWithUTF8String:key];
+ return static_cast<int>([defaults integerForKey:ns_key]);
+ };
+
+ g_rpc_args.host_url = get_string_from_cache("tmvrpc_url");
+ g_rpc_args.host_port = get_int_from_cache("tmvrpc_port");
+ g_rpc_args.key = get_string_from_cache("tmvrpc_key");
+}
+
+static void update_in_cache() {
+ NSUserDefaults* defaults = [NSUserDefaults standardUserDefaults];
+
+ [defaults setObject:[NSString stringWithUTF8String:g_rpc_args.host_url.c_str()]
+ forKey:@"tmvrpc_url"];
+ [defaults setInteger:g_rpc_args.host_port forKey:@"tmvrpc_port"];
+ [defaults setObject:[NSString stringWithUTF8String:g_rpc_args.key.c_str()] forKey:@"tmvrpc_key"];
Review comment:
nit: tvm, here and above
##########
File path: apps/ios_rpc/tvmrpc/RPCServer.mm
##########
@@ -0,0 +1,809 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*!
+ * \file ViewController.mm
+ */
+
+#import "RPCServer.h"
+
+#include <tvm/runtime/packed_func.h>
+#include <tvm/runtime/registry.h>
+
+#include <random>
+#include <string>
+
+// To get device WiFi IP
+#include <arpa/inet.h>
+#include <ifaddrs.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+
+// TVM internal header to access Magic keys like kRPCMagic and others
+#include "../../../src/runtime/rpc/rpc_endpoint.h"
+
+namespace tvm {
+namespace runtime {
+
+/*!
+ * \brief Message handling function for event driven server.
+ *
+ * \param in_bytes The incoming bytes.
+ * \param event_flag 1: read_available, 2: write_avaiable.
+ * \return State flag.
+ * 1: continue running, no need to write,
+ * 2: need to write
+ * 0: shutdown
+ */
+using FEventHandler = PackedFunc;
+
+/*!
+ * \brief Create a server event handler.
+ *
+ * \param outputStream The output stream used to send outputs.
+ * \param name The name of the server.
+ * \param remote_key The remote key
+ * \return The event handler.
+ */
+FEventHandler CreateServerEventHandler(NSOutputStream* outputStream, std::string name,
+ std::string remote_key) {
+ const PackedFunc* event_handler_factor = Registry::Get("rpc.CreateEventDrivenServer");
+ ICHECK(event_handler_factor != nullptr)
+ << "You are using tvm_runtime module built without RPC support. "
+ << "Please rebuild it with USE_RPC flag.";
+
+ PackedFunc writer_func([outputStream](TVMArgs args, TVMRetValue* rv) {
+ TVMByteArray* data = args[0].ptr<TVMByteArray>();
+ int64_t nbytes = [outputStream write:reinterpret_cast<const uint8_t*>(data->data)
+ maxLength:data->size];
+ if (nbytes < 0) {
+ NSLog(@"%@", [outputStream streamError].localizedDescription);
+ throw tvm::Error("Stream error");
+ }
+ *rv = nbytes;
+ });
+
+ return (*event_handler_factor)(writer_func, name, remote_key);
+}
+
+/*!
+ * \brief Helper function to query real IP of device in WiFi network
+ * \return string with IPv4 in format "192.168.0.1" or "unknown" if cannot detect
+ */
+static std::string getWiFiAddress() {
+ std::string address = "unknown";
+ ifaddrs* interfaces = nullptr;
+
+ int success = getifaddrs(&interfaces);
+ if (success == 0) {
+ ifaddrs* temp_addr = interfaces;
+ while (temp_addr != NULL) {
+ if (temp_addr->ifa_addr->sa_family == AF_INET) {
+ // Check if interface is en0 which is the wifi connection on the iPhone
+ if (std::string(temp_addr->ifa_name) == "en0") {
+ address = inet_ntoa(((sockaddr_in*)temp_addr->ifa_addr)->sin_addr);
+ }
+ }
+ temp_addr = temp_addr->ifa_next;
+ }
+ }
+
+ freeifaddrs(interfaces);
+ return address;
+}
+
+} // namespace runtime
+} // namespace tvm
+
+// Base class for any type of RPC servicing
+@interface RPCServerBase : RPCServer
+
+/*!
+ * Methods should be implemented in inherited classes
+ */
+- (bool)onReadHandler; // return true - continue feeding, false - stop, try to drain output buffer
+- (bool)onWriteHandler; // return true - continue draining, false - no data to write
+- (void)onEndEncountered; // called on disconnect or session desided that it's shutdown time
+- (void)open; // Initiate listening objects like i/o streams and other resources
+- (void)close; // Deinitialize resources opend in "open" method
+@end
+
+@implementation RPCServerBase {
+ // Worker thread
+ NSThread* worker_thread_;
+ // Triger to continue RunLoop processing inside worker_thread_
+ BOOL shouldKeepRunning;
+ // Input socket stream
+ @protected
+ NSInputStream* inputStream_;
+ // Output socket stream
+ NSOutputStream* outputStream_;
+ // Temporal buffer with data to send
+ std::string sendBuffer_;
+ // Temporal receive buffer
+ std::string recvBuffer_;
+ // Requested data size to accumulate in recvBuffer_ before continue processing
+ int requiredToRecv_;
+}
+
+/*!
+ * Start internal worker thread with RunLoop and submit correspoding open handlers into it
+ * Not blocking
+ */
+- (void)start {
+ worker_thread_ = [[NSThread alloc] initWithBlock:^{
+ @autoreleasepool {
+ [self notifyState:RPCServerStatus_Launched];
+ [self open];
+ shouldKeepRunning = YES;
+ while (shouldKeepRunning && [[NSRunLoop currentRunLoop] runMode:NSDefaultRunLoopMode
+ beforeDate:[NSDate distantFuture]])
+ ;
+ [self notifyState:RPCServerStatus_Stopped];
+ }
+ }];
+ [worker_thread_ start];
+}
+
+/*!
+ * Send message to workel thread runloop to finish processing
+ * Not blocking
+ */
+- (void)stop {
+ if (worker_thread_ == nil) return;
+
+ [self performSelector:@selector(stop_) onThread:worker_thread_ withObject:nil waitUntilDone:NO];
+ worker_thread_ = nil; // TODO: is it valide? may be better to do that inside NSThread?
+}
+
+- (void)stop_ {
+ [self close];
+ shouldKeepRunning = NO;
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will connect to host and port specified in corresponding properties
+ */
+- (void)open {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)self.host, self.port, &readStream,
+ &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will assign i/o streams to provided socket connection.
+ */
+- (void)openWithSocket:(CFSocketNativeHandle)sock {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocket(NULL, sock, &readStream, &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Close i/o streams assosiated with connection
+ */
+- (void)close {
+ [inputStream_ close];
+ [outputStream_ close];
+ [inputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [inputStream_ setDelegate:nil];
+ [outputStream_ setDelegate:nil];
+ inputStream_ = nil;
+ outputStream_ = nil;
+}
+
+/// Unimplemented stubs
+- (bool)onReadHandler {
+ return false;
+}
+- (bool)onWriteHandler {
+ return false;
+}
+- (void)onEndEncountered {
+}
+
+/*!
+ * Try to read data from stream and call processing hadnler
+ */
+- (void)tryToRead {
+ const int kBufferSize = 4 << 10; // 4kB buffer
+ const int prev_size = recvBuffer_.size();
+ recvBuffer_.resize(kBufferSize);
+ size_t nbytes = [inputStream_ read:(uint8_t*)recvBuffer_.data() + prev_size
+ maxLength:recvBuffer_.size() - prev_size];
+ recvBuffer_.resize(nbytes + prev_size);
+
+ // feed while it accept or requested particulat buffer size
+ while (!recvBuffer_.empty() && requiredToRecv_ <= recvBuffer_.size() && [self onReadHandler])
+ ;
+}
+
+/*!
+ * Try to write remaining data to stream and call processing hadnler
+ */
+- (void)tryToWrite {
+ if (!sendBuffer_.empty()) {
+ size_t nbytes = [outputStream_ write:(uint8_t*)sendBuffer_.data() maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, nbytes);
+ }
+ // call write handler while it want be called and space is available
+ while (sendBuffer_.empty() && [outputStream_ hasSpaceAvailable] && [self onWriteHandler])
+ ;
+}
+
+/*!
+ * Main event handler of socket stream events
+ */
+- (void)stream:(NSStream*)strm handleEvent:(NSStreamEvent)event {
+ std::string buffer;
+ switch (event) {
+ case NSStreamEventOpenCompleted: {
+ // Nothing
+ break;
+ }
+ case NSStreamEventHasBytesAvailable:
+ if (strm == inputStream_) {
+ [self tryToRead];
+ if ([outputStream_ hasSpaceAvailable]) [self tryToWrite];
+ }
+ break;
+ case NSStreamEventHasSpaceAvailable: {
+ if (strm == outputStream_) {
+ [self tryToWrite];
+ if ([inputStream_ hasBytesAvailable]) [self tryToRead];
+ }
+ break;
+ }
+ case NSStreamEventErrorOccurred: {
+ [self notifyError:[strm streamError].localizedDescription];
+ break;
+ }
+ case NSStreamEventEndEncountered: {
+ [self onEndEncountered];
+ break;
+ }
+ default: {
+ NSLog(@"Unknown event");
+ }
+ }
+}
+
+#pragma mark - Helpers
+
+/*!
+ * Set buffer to send into stream. Try to send immediatly or submit to lazy sending
+ * Non blocking operation
+ */
+- (void)toSend:(NSData*)data {
+ sendBuffer_.append(static_cast<const char*>(data.bytes), data.length);
+
+ // try to flush buffer
+ NSInteger sent_size = [outputStream_ write:(uint8_t*)sendBuffer_.data()
+ maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, sent_size);
+}
+
+/*!
+ * Set buffer to send in packet format [size, data]. Behaviour is same as for toSend.
+ */
+- (void)toSendPacked:(NSData*)data {
+ int packet_size = data.length;
+ [self toSend:[NSData dataWithBytes:&packet_size length:sizeof(packet_size)]];
+ [self toSend:data];
+}
+
+/*!
+ */
+- (NSData*)requestInputDataWithSize:(NSInteger)size {
+ if (recvBuffer_.size() < size) {
+ requiredToRecv_ = size;
+ return nil;
+ }
+ NSData* res = [NSData dataWithBytes:recvBuffer_.data() length:size];
+ recvBuffer_.erase(0, size);
+ return res;
+}
+
+/*!
+ */
+- (NSData*)requestInputDataPacked {
+ int size;
Review comment:
prefer to use the inttypes when taking sizeof() to compute payload lengths for wire data
##########
File path: apps/ios_rpc/tvmrpc/RPCArgs.mm
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#import "RPCArgs.h"
+
+#import <Foundation/Foundation.h>
+
+#import "../../../src/support/socket.h"
+#import "../../../src/support/utils.h"
+
+#import <string>
+
+using std::string;
+
+const char* kUsage =
+ "\n"
+ "iOS tvmrpc application supported flags:\n"
+ "--host_url The tracker/proxy address, Default=0.0.0.0\n"
+ "--host_port The tracker/proxy port, Default=9190\n"
+ "--key The key used to identify the device type in tracker. Default=\"\"\n"
+ "--custom_addr Custom IP Address to Report to RPC Tracker. Default=\"\"\n"
+ "--immediate_connect No UI interconnection, connect to tracker immediately. Default=False\n"
+ "--verbose Allow to print status info to std out. Default=False\n"
+ "--server_mode Server mode. Can be \"pure_server\", \"proxy\" or \"tracker\". "
+ "Default=pure_server \n"
+ "\n";
+
+struct RPCArgs_cpp {
+ string host_url = "0.0.0.0";
+ int host_port = 9190;
+
+ string key;
+ string custom_addr = "";
+
+ bool immediate_connect = false;
+ bool verbose = false;
+ char server_mode = 0;
+
+ operator RPCArgs() const {
+ return RPCArgs{.host_url = host_url.c_str(),
+ .host_port = host_port,
+ .key = key.c_str(),
+ .custom_addr = custom_addr.c_str(),
+ .verbose = verbose,
+ .immediate_connect = immediate_connect,
+ .server_mode = server_mode};
+ };
+
+ RPCArgs_cpp& operator=(const RPCArgs& args) {
+ host_url = args.host_url;
+ host_port = args.host_port;
+ key = args.key;
+ custom_addr = args.custom_addr;
+ verbose = args.verbose;
+ immediate_connect = args.immediate_connect;
+ server_mode = args.server_mode;
+ return *this;
+ }
+};
+
+struct RPCArgs_cpp g_rpc_args;
+
+static void restore_from_cache() {
+ NSUserDefaults* defaults = [NSUserDefaults standardUserDefaults];
+
+ auto get_string_from_cache = [defaults](const char* key) {
+ NSString* ns_key = [NSString stringWithUTF8String:key];
+ NSString* ns_val = [defaults stringForKey:ns_key];
+ return std::string(ns_val != nil ? [ns_val UTF8String] : "");
+ };
+
+ auto get_int_from_cache = [defaults](const char* key) {
+ NSString* ns_key = [NSString stringWithUTF8String:key];
+ return static_cast<int>([defaults integerForKey:ns_key]);
+ };
+
+ g_rpc_args.host_url = get_string_from_cache("tmvrpc_url");
+ g_rpc_args.host_port = get_int_from_cache("tmvrpc_port");
+ g_rpc_args.key = get_string_from_cache("tmvrpc_key");
+}
+
+static void update_in_cache() {
+ NSUserDefaults* defaults = [NSUserDefaults standardUserDefaults];
+
+ [defaults setObject:[NSString stringWithUTF8String:g_rpc_args.host_url.c_str()]
+ forKey:@"tmvrpc_url"];
+ [defaults setInteger:g_rpc_args.host_port forKey:@"tmvrpc_port"];
+ [defaults setObject:[NSString stringWithUTF8String:g_rpc_args.key.c_str()] forKey:@"tmvrpc_key"];
+}
+
+string GetCmdOption(int argc, char* argv[], string option, bool key = false) {
+ string cmd;
+ for (int i = 1; i < argc; ++i) {
+ string arg = argv[i];
+ if (arg.find(option) == 0) {
+ if (key) {
+ cmd = argv[i];
+ return cmd;
+ }
+ // We assume "=" is the end of option.
+ ICHECK_EQ(*option.rbegin(), '=');
+ cmd = arg.substr(arg.find('=') + 1);
+ return cmd;
+ }
+ }
+ return cmd;
+}
+
+void update_rpc_args(int argc, char* argv[]) {
+ restore_from_cache();
+ RPCArgs_cpp& args = g_rpc_args;
+
+ using tvm::support::IsNumber;
+ using tvm::support::ValidateIP;
+ constexpr int MAX_PORT_NUM = 65535;
+
+ const string immediate_connect = GetCmdOption(argc, argv, "--immediate_connect", true);
+ args.immediate_connect = !immediate_connect.empty();
+
+ const string verbose = GetCmdOption(argc, argv, "--verbose", true);
+ args.verbose = !verbose.empty();
+
+ const string server_mode = GetCmdOption(argc, argv, "--server_mode=", false);
+ if (!server_mode.empty()) {
+ if (server_mode == "tracker") {
+ args.server_mode = 0;
+ } else if (server_mode == "proxy") {
+ args.server_mode = 1;
+ } else if (server_mode == "pure_server") {
Review comment:
what's the use case for gRPC?
##########
File path: apps/ios_rpc/tvmrpc/RPCArgs.h
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef TVM_APPS_IOS_RPC_ARGS_H_
+#define TVM_APPS_IOS_RPC_ARGS_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/*!
+ * \brief Struct representing arguments of iOS RPC app
+ */
+typedef struct RPCArgs_t {
+ /// Tracker or Proxy address (actually ip)
+ const char* host_url;
+
+ /// Tracker or Proxy port
+ int host_port;
+
+ /// device key to report
+ const char* key;
+
+ /// custom adress to report into Tracker. Ignored for other server modes.
+ const char* custom_addr;
+
+ /// Verbose mode. Will print status messages to std out.
+ /// 0 - no prints , 1 - print state to output
+ char verbose;
+
+ /// Immediate server launch. No UI interaction.
+ /// 0 - UI interaction, 1 - automatically connect on launch
+ char immediate_connect;
Review comment:
why do you represent bool quantities with char? (instead of bool or uint8_t)
##########
File path: apps/ios_rpc/tvmrpc/RPCArgs.mm
##########
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#import "RPCArgs.h"
+
+#import <Foundation/Foundation.h>
+
+#import "../../../src/support/socket.h"
+#import "../../../src/support/utils.h"
+
+#import <string>
+
+using std::string;
+
+const char* kUsage =
+ "\n"
+ "iOS tvmrpc application supported flags:\n"
+ "--host_url The tracker/proxy address, Default=0.0.0.0\n"
+ "--host_port The tracker/proxy port, Default=9190\n"
+ "--key The key used to identify the device type in tracker. Default=\"\"\n"
+ "--custom_addr Custom IP Address to Report to RPC Tracker. Default=\"\"\n"
+ "--immediate_connect No UI interconnection, connect to tracker immediately. Default=False\n"
+ "--verbose Allow to print status info to std out. Default=False\n"
+ "--server_mode Server mode. Can be \"pure_server\", \"proxy\" or \"tracker\". "
+ "Default=pure_server \n"
+ "\n";
+
+struct RPCArgs_cpp {
+ string host_url = "0.0.0.0";
+ int host_port = 9190;
+
+ string key;
+ string custom_addr = "";
+
+ bool immediate_connect = false;
+ bool verbose = false;
+ char server_mode = 0;
Review comment:
should that be an enum?
##########
File path: apps/ios_rpc/tvmrpc/RPCServer.mm
##########
@@ -0,0 +1,809 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*!
+ * \file ViewController.mm
+ */
+
+#import "RPCServer.h"
+
+#include <tvm/runtime/packed_func.h>
+#include <tvm/runtime/registry.h>
+
+#include <random>
+#include <string>
+
+// To get device WiFi IP
+#include <arpa/inet.h>
+#include <ifaddrs.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+
+// TVM internal header to access Magic keys like kRPCMagic and others
+#include "../../../src/runtime/rpc/rpc_endpoint.h"
+
+namespace tvm {
+namespace runtime {
+
+/*!
+ * \brief Message handling function for event driven server.
+ *
+ * \param in_bytes The incoming bytes.
+ * \param event_flag 1: read_available, 2: write_avaiable.
+ * \return State flag.
+ * 1: continue running, no need to write,
+ * 2: need to write
+ * 0: shutdown
+ */
+using FEventHandler = PackedFunc;
+
+/*!
+ * \brief Create a server event handler.
+ *
+ * \param outputStream The output stream used to send outputs.
+ * \param name The name of the server.
+ * \param remote_key The remote key
+ * \return The event handler.
+ */
+FEventHandler CreateServerEventHandler(NSOutputStream* outputStream, std::string name,
+ std::string remote_key) {
+ const PackedFunc* event_handler_factor = Registry::Get("rpc.CreateEventDrivenServer");
+ ICHECK(event_handler_factor != nullptr)
+ << "You are using tvm_runtime module built without RPC support. "
+ << "Please rebuild it with USE_RPC flag.";
+
+ PackedFunc writer_func([outputStream](TVMArgs args, TVMRetValue* rv) {
+ TVMByteArray* data = args[0].ptr<TVMByteArray>();
+ int64_t nbytes = [outputStream write:reinterpret_cast<const uint8_t*>(data->data)
+ maxLength:data->size];
+ if (nbytes < 0) {
+ NSLog(@"%@", [outputStream streamError].localizedDescription);
+ throw tvm::Error("Stream error");
+ }
+ *rv = nbytes;
+ });
+
+ return (*event_handler_factor)(writer_func, name, remote_key);
+}
+
+/*!
+ * \brief Helper function to query real IP of device in WiFi network
+ * \return string with IPv4 in format "192.168.0.1" or "unknown" if cannot detect
+ */
+static std::string getWiFiAddress() {
+ std::string address = "unknown";
+ ifaddrs* interfaces = nullptr;
+
+ int success = getifaddrs(&interfaces);
+ if (success == 0) {
+ ifaddrs* temp_addr = interfaces;
+ while (temp_addr != NULL) {
+ if (temp_addr->ifa_addr->sa_family == AF_INET) {
+ // Check if interface is en0 which is the wifi connection on the iPhone
+ if (std::string(temp_addr->ifa_name) == "en0") {
+ address = inet_ntoa(((sockaddr_in*)temp_addr->ifa_addr)->sin_addr);
+ }
+ }
+ temp_addr = temp_addr->ifa_next;
+ }
+ }
+
+ freeifaddrs(interfaces);
+ return address;
+}
+
+} // namespace runtime
+} // namespace tvm
+
+// Base class for any type of RPC servicing
+@interface RPCServerBase : RPCServer
+
+/*!
+ * Methods should be implemented in inherited classes
+ */
+- (bool)onReadHandler; // return true - continue feeding, false - stop, try to drain output buffer
+- (bool)onWriteHandler; // return true - continue draining, false - no data to write
+- (void)onEndEncountered; // called on disconnect or session desided that it's shutdown time
+- (void)open; // Initiate listening objects like i/o streams and other resources
+- (void)close; // Deinitialize resources opend in "open" method
+@end
+
+@implementation RPCServerBase {
+ // Worker thread
+ NSThread* worker_thread_;
+ // Triger to continue RunLoop processing inside worker_thread_
+ BOOL shouldKeepRunning;
+ // Input socket stream
+ @protected
+ NSInputStream* inputStream_;
+ // Output socket stream
+ NSOutputStream* outputStream_;
+ // Temporal buffer with data to send
+ std::string sendBuffer_;
+ // Temporal receive buffer
+ std::string recvBuffer_;
+ // Requested data size to accumulate in recvBuffer_ before continue processing
+ int requiredToRecv_;
+}
+
+/*!
+ * Start internal worker thread with RunLoop and submit correspoding open handlers into it
+ * Not blocking
+ */
+- (void)start {
+ worker_thread_ = [[NSThread alloc] initWithBlock:^{
+ @autoreleasepool {
+ [self notifyState:RPCServerStatus_Launched];
+ [self open];
+ shouldKeepRunning = YES;
+ while (shouldKeepRunning && [[NSRunLoop currentRunLoop] runMode:NSDefaultRunLoopMode
+ beforeDate:[NSDate distantFuture]])
+ ;
+ [self notifyState:RPCServerStatus_Stopped];
+ }
+ }];
+ [worker_thread_ start];
+}
+
+/*!
+ * Send message to workel thread runloop to finish processing
+ * Not blocking
+ */
+- (void)stop {
+ if (worker_thread_ == nil) return;
+
+ [self performSelector:@selector(stop_) onThread:worker_thread_ withObject:nil waitUntilDone:NO];
+ worker_thread_ = nil; // TODO: is it valide? may be better to do that inside NSThread?
+}
+
+- (void)stop_ {
+ [self close];
+ shouldKeepRunning = NO;
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will connect to host and port specified in corresponding properties
+ */
+- (void)open {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)self.host, self.port, &readStream,
+ &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will assign i/o streams to provided socket connection.
+ */
+- (void)openWithSocket:(CFSocketNativeHandle)sock {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocket(NULL, sock, &readStream, &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Close i/o streams assosiated with connection
+ */
+- (void)close {
+ [inputStream_ close];
+ [outputStream_ close];
+ [inputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [inputStream_ setDelegate:nil];
+ [outputStream_ setDelegate:nil];
+ inputStream_ = nil;
+ outputStream_ = nil;
+}
+
+/// Unimplemented stubs
+- (bool)onReadHandler {
+ return false;
+}
+- (bool)onWriteHandler {
+ return false;
+}
+- (void)onEndEncountered {
+}
+
+/*!
+ * Try to read data from stream and call processing hadnler
+ */
+- (void)tryToRead {
+ const int kBufferSize = 4 << 10; // 4kB buffer
+ const int prev_size = recvBuffer_.size();
+ recvBuffer_.resize(kBufferSize);
+ size_t nbytes = [inputStream_ read:(uint8_t*)recvBuffer_.data() + prev_size
+ maxLength:recvBuffer_.size() - prev_size];
+ recvBuffer_.resize(nbytes + prev_size);
+
+ // feed while it accept or requested particulat buffer size
+ while (!recvBuffer_.empty() && requiredToRecv_ <= recvBuffer_.size() && [self onReadHandler])
+ ;
+}
+
+/*!
+ * Try to write remaining data to stream and call processing hadnler
+ */
+- (void)tryToWrite {
+ if (!sendBuffer_.empty()) {
+ size_t nbytes = [outputStream_ write:(uint8_t*)sendBuffer_.data() maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, nbytes);
+ }
+ // call write handler while it want be called and space is available
+ while (sendBuffer_.empty() && [outputStream_ hasSpaceAvailable] && [self onWriteHandler])
+ ;
+}
+
+/*!
+ * Main event handler of socket stream events
+ */
+- (void)stream:(NSStream*)strm handleEvent:(NSStreamEvent)event {
+ std::string buffer;
+ switch (event) {
+ case NSStreamEventOpenCompleted: {
+ // Nothing
+ break;
+ }
+ case NSStreamEventHasBytesAvailable:
+ if (strm == inputStream_) {
+ [self tryToRead];
+ if ([outputStream_ hasSpaceAvailable]) [self tryToWrite];
+ }
+ break;
+ case NSStreamEventHasSpaceAvailable: {
+ if (strm == outputStream_) {
+ [self tryToWrite];
+ if ([inputStream_ hasBytesAvailable]) [self tryToRead];
+ }
+ break;
+ }
+ case NSStreamEventErrorOccurred: {
+ [self notifyError:[strm streamError].localizedDescription];
+ break;
+ }
+ case NSStreamEventEndEncountered: {
+ [self onEndEncountered];
+ break;
+ }
+ default: {
+ NSLog(@"Unknown event");
+ }
+ }
+}
+
+#pragma mark - Helpers
+
+/*!
+ * Set buffer to send into stream. Try to send immediatly or submit to lazy sending
+ * Non blocking operation
+ */
+- (void)toSend:(NSData*)data {
+ sendBuffer_.append(static_cast<const char*>(data.bytes), data.length);
+
+ // try to flush buffer
+ NSInteger sent_size = [outputStream_ write:(uint8_t*)sendBuffer_.data()
+ maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, sent_size);
+}
+
+/*!
+ * Set buffer to send in packet format [size, data]. Behaviour is same as for toSend.
+ */
+- (void)toSendPacked:(NSData*)data {
+ int packet_size = data.length;
+ [self toSend:[NSData dataWithBytes:&packet_size length:sizeof(packet_size)]];
+ [self toSend:data];
+}
+
+/*!
+ */
+- (NSData*)requestInputDataWithSize:(NSInteger)size {
+ if (recvBuffer_.size() < size) {
+ requiredToRecv_ = size;
+ return nil;
+ }
+ NSData* res = [NSData dataWithBytes:recvBuffer_.data() length:size];
+ recvBuffer_.erase(0, size);
+ return res;
+}
+
+/*!
+ */
+- (NSData*)requestInputDataPacked {
+ int size;
+ if (recvBuffer_.size() < sizeof(size)) {
+ requiredToRecv_ = sizeof(size);
+ return nil;
+ }
+ size = *(int*)recvBuffer_.data();
+ if (recvBuffer_.size() < sizeof(size) + size) {
+ requiredToRecv_ = sizeof(size) + size;
+ return nil;
+ }
+ NSData* res = [NSData dataWithBytes:recvBuffer_.data() + sizeof(size) length:size];
+ recvBuffer_.erase(0, sizeof(size) + size);
+ return res;
+};
+
+#pragma mark - Notifiers
+
+/*!
+ * Notify external listener about error.
+ * Also print error message to std out in case of Verbose mode
+ */
+- (void)notifyError:(NSString*)msg {
+ // Duplicate error message in std output. Host launcher script may listen it.
+ if (self.verbose) NSLog(@"[IOS-RPC] ERROR: %@", msg);
+ if (self.delegate) [self.delegate onError:msg];
+}
+
+/*!
+ * Notify external listener about server state changes.
+ * Also print information to std out in case of Verbose mode
+ */
+- (void)notifyState:(RPCServerStatus)state {
+ // Duplicate sattus changing in std output. Host launcher script may listen it.
+ if (self.verbose) NSLog(@"[IOS-RPC] STATE: %d", state);
+ if (self.delegate != nil) [self.delegate onStatusChanged:state];
+}
+
+@end
+
+@interface RPCServerProxy : RPCServerBase
+@end
+
+typedef enum {
+ RPCServerProxyState_Idle,
+ RPCServerProxyState_HandshakeToSend,
+ RPCServerProxyState_HandshakeToRecv,
+ RPCServerProxyState_Processing,
+} RPCServerProxyState;
+
+@implementation RPCServerProxy {
+ /// Original TVM RPC event handler
+ tvm::runtime::FEventHandler handler_;
+ @protected
+ /// Sate of Proxy client implementation
+ RPCServerProxyState state_;
+}
+
+- (instancetype)init {
+ if (self = [super init]) {
+ handler_ = nullptr;
+ state_ = RPCServerProxyState_Idle;
+ }
+ return self;
+}
+
+/*!
+ * Implement matching of internat state on state available for outside users
+ */
+- (void)setState:(RPCServerProxyState)new_state {
+ // Send Connected notification because Proxy doesn't responce until client connected.
+ if (new_state == RPCServerProxyState_HandshakeToRecv)
+ [self notifyState:RPCServerStatus_Connected];
+ if (new_state == RPCServerProxyState_Idle) [self notifyState:RPCServerStatus_Disconnected];
+ if (state_ == RPCServerProxyState_HandshakeToRecv && new_state == RPCServerProxyState_Processing)
+ [self notifyState:RPCServerStatus_RPCSessionStarted];
+ if (state_ == RPCServerProxyState_Processing && new_state == RPCServerProxyState_Idle)
+ [self notifyState:RPCServerStatus_RPCSessionStarted];
+
+ state_ = new_state;
+}
+
+- (bool)onWriteHandler {
+ switch (state_) {
+ case RPCServerProxyState_HandshakeToSend: {
+ // Send together kRPCMagic and server descriptor because of Proxy
+ int code = tvm::runtime::kRPCMagic;
Review comment:
inttypes
##########
File path: apps/ios_rpc/tvmrpc/ViewController.mm
##########
@@ -22,168 +22,144 @@
*/
#import "ViewController.h"
-#include <string>
+#import "RPCArgs.h"
-@implementation ViewController
-
-- (void)stream:(NSStream*)strm handleEvent:(NSStreamEvent)event {
- std::string buffer;
- switch (event) {
- case NSStreamEventOpenCompleted: {
- self.statusLabel.text = @"Connected";
- break;
- }
- case NSStreamEventHasBytesAvailable:
- if (strm == inputStream_) {
- [self onReadAvailable];
- }
- break;
- case NSStreamEventHasSpaceAvailable: {
- if (strm == outputStream_) {
- [self onWriteAvailable];
- }
- break;
- }
- case NSStreamEventErrorOccurred: {
- NSLog(@"%@", [strm streamError].localizedDescription);
- break;
- }
- case NSStreamEventEndEncountered: {
- [self close];
- // auto reconnect when normal end.
- [self open];
- break;
- }
- default: {
- NSLog(@"Unknown event");
- }
- }
+@implementation ViewController {
+ // server implementation
+ RPCServer* server_;
+ // verbose flag to print status info
+ bool verbose_;
+ // Button state. True - push will start connection, false - push will disconnect
+ bool to_connect_;
}
-- (void)onReadAvailable {
- constexpr int kRPCMagic = 0xff271;
- if (!initialized_) {
- int code;
- size_t nbytes = [inputStream_ read:reinterpret_cast<uint8_t*>(&code) maxLength:sizeof(code)];
- if (nbytes != sizeof(code)) {
- self.infoText.text = @"Fail to receive remote confirmation code.";
- [self close];
- } else if (code == kRPCMagic + 2) {
- self.infoText.text = @"Proxy server cannot find client that matches the key";
- [self close];
- } else if (code == kRPCMagic + 1) {
- self.infoText.text = @"Proxy server already have another server with same key";
- [self close];
- } else if (code != kRPCMagic) {
- self.infoText.text = @"Given address is not a TVM RPC Proxy";
- [self close];
- } else {
- initialized_ = true;
- self.statusLabel.text = @"Proxy connected.";
- ICHECK(handler_ != nullptr);
- }
- }
- const int kBufferSize = 4 << 10;
- if (initialized_) {
- while ([inputStream_ hasBytesAvailable]) {
- recvBuffer_.resize(kBufferSize);
- uint8_t* bptr = reinterpret_cast<uint8_t*>(&recvBuffer_[0]);
- size_t nbytes = [inputStream_ read:bptr maxLength:kBufferSize];
- recvBuffer_.resize(nbytes);
- int flag = 1;
- if ([outputStream_ hasSpaceAvailable]) {
- flag |= 2;
- }
- // always try to write
- try {
- flag = handler_(recvBuffer_, flag);
- if (flag == 2) {
- [self onShutdownReceived];
- }
- } catch (const tvm::Error& e) {
- [self close];
- }
- }
+- (void)viewDidLoad {
+ // To handle end editing events
+ self.proxyURL.delegate = self;
+ self.proxyPort.delegate = self;
+ self.proxyKey.delegate = self;
+
+ RPCArgs args = get_current_rpc_args();
+ self.proxyURL.text = @(args.host_url);
+ self.proxyPort.text = @(args.host_port).stringValue;
+ self.proxyKey.text = @(args.key);
+
+ self.ModeSelector.selectedSegmentIndex = args.server_mode;
+ self->verbose_ = args.verbose;
+ self->to_connect_ = true;
+
+ // Add border to button
+ void (^addBorder)(UIButton* btn) = ^(UIButton* btn) {
+ btn.layer.borderWidth = 2.0f;
+ btn.layer.borderColor = self.ConnectButton.currentTitleColor.CGColor;
+ btn.layer.cornerRadius = 10;
+ };
+ addBorder(self.ConnectButton);
+
+ // Connect to tracker immediately
+ if (args.immediate_connect) {
+ [self disableUIInteraction];
+ [self open];
}
}
-- (void)onShutdownReceived {
- [self close];
-}
+/*!
+ * \brief Disable all UI elements
+ */
+- (void)disableUIInteraction {
+ void (^disable)(UITextField* field) = ^(UITextField* field) {
+ field.enabled = NO;
+ field.backgroundColor = [UIColor lightGrayColor];
+ };
-- (void)onWriteAvailable {
- if (initSendPtr_ < initBytes_.length()) {
- initSendPtr_ += [outputStream_ write:reinterpret_cast<uint8_t*>(&initBytes_[initSendPtr_])
- maxLength:(initBytes_.length() - initSendPtr_)];
- }
- if (initialized_) {
- try {
- std::string dummy;
- int flag = handler_(dummy, 2);
- if (flag == 2) {
- [self onShutdownReceived];
- }
- } catch (const tvm::Error& e) {
- [self close];
- }
- }
+ void (^disableButton)(UIButton* btn) = ^(UIButton* btn) {
+ btn.enabled = NO;
+ btn.layer.borderColor = btn.currentTitleColor.CGColor;
+ };
+
+ disable(self.proxyURL);
+ disable(self.proxyPort);
+ disable(self.proxyKey);
+ disableButton(self.ConnectButton);
+ self.ModeSelector.enabled = NO;
}
+/*!
+ * \brief Start RPC server
+ */
- (void)open {
- constexpr int kRPCMagic = 0xff271;
- NSLog(@"Connecting to the proxy server..");
- // Initialize the data states.
- key_ = [self.proxyKey.text UTF8String];
- key_ = "server:" + key_;
- std::ostringstream os;
- int rpc_magic = kRPCMagic;
- os.write(reinterpret_cast<char*>(&rpc_magic), sizeof(rpc_magic));
- int keylen = static_cast<int>(key_.length());
- os.write(reinterpret_cast<char*>(&keylen), sizeof(keylen));
- os.write(key_.c_str(), key_.length());
- initialized_ = false;
- initBytes_ = os.str();
- initSendPtr_ = 0;
- // Initialize the network.
- CFReadStreamRef readStream;
- CFWriteStreamRef writeStream;
- CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)self.proxyURL.text,
- [self.proxyPort.text intValue], &readStream, &writeStream);
- inputStream_ = (NSInputStream*)readStream;
- outputStream_ = (NSOutputStream*)writeStream;
- [inputStream_ setDelegate:self];
- [outputStream_ setDelegate:self];
- [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
- [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
- [outputStream_ open];
- [inputStream_ open];
- handler_ = tvm::runtime::CreateServerEventHandler(outputStream_, key_, "%toinit");
- ICHECK(handler_ != nullptr);
+ RPCServerMode server_mode = static_cast<RPCServerMode>(self.ModeSelector.selectedSegmentIndex);
+
+ server_ = [RPCServer serverWithMode:server_mode];
+ server_.host = self.proxyURL.text;
+ server_.port = self.proxyPort.text.intValue;
+ server_.key = self.proxyKey.text;
+ server_.verbose = self->verbose_;
+ server_.delegate = self;
+
+ [server_ start];
+
self.infoText.text = @"";
self.statusLabel.text = @"Connecting...";
}
+/*!
+ * \brief Stop RPC server
+ */
- (void)close {
- NSLog(@"Closing the streams.");
- [inputStream_ close];
- [outputStream_ close];
- [inputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
- [outputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
- [inputStream_ setDelegate:nil];
- [outputStream_ setDelegate:nil];
- inputStream_ = nil;
- outputStream_ = nil;
- handler_ = nullptr;
- self.statusLabel.text = @"Disconnected";
+ [server_ stop];
+ self.statusLabel.text = @"Disconnecting...";
Review comment:
maybe swap this with previous line?
##########
File path: apps/ios_rpc/tvmrpc/RPCServer.mm
##########
@@ -0,0 +1,809 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/*!
+ * \file ViewController.mm
+ */
+
+#import "RPCServer.h"
+
+#include <tvm/runtime/packed_func.h>
+#include <tvm/runtime/registry.h>
+
+#include <random>
+#include <string>
+
+// To get device WiFi IP
+#include <arpa/inet.h>
+#include <ifaddrs.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+
+// TVM internal header to access Magic keys like kRPCMagic and others
+#include "../../../src/runtime/rpc/rpc_endpoint.h"
+
+namespace tvm {
+namespace runtime {
+
+/*!
+ * \brief Message handling function for event driven server.
+ *
+ * \param in_bytes The incoming bytes.
+ * \param event_flag 1: read_available, 2: write_avaiable.
+ * \return State flag.
+ * 1: continue running, no need to write,
+ * 2: need to write
+ * 0: shutdown
+ */
+using FEventHandler = PackedFunc;
+
+/*!
+ * \brief Create a server event handler.
+ *
+ * \param outputStream The output stream used to send outputs.
+ * \param name The name of the server.
+ * \param remote_key The remote key
+ * \return The event handler.
+ */
+FEventHandler CreateServerEventHandler(NSOutputStream* outputStream, std::string name,
+ std::string remote_key) {
+ const PackedFunc* event_handler_factor = Registry::Get("rpc.CreateEventDrivenServer");
+ ICHECK(event_handler_factor != nullptr)
+ << "You are using tvm_runtime module built without RPC support. "
+ << "Please rebuild it with USE_RPC flag.";
+
+ PackedFunc writer_func([outputStream](TVMArgs args, TVMRetValue* rv) {
+ TVMByteArray* data = args[0].ptr<TVMByteArray>();
+ int64_t nbytes = [outputStream write:reinterpret_cast<const uint8_t*>(data->data)
+ maxLength:data->size];
+ if (nbytes < 0) {
+ NSLog(@"%@", [outputStream streamError].localizedDescription);
+ throw tvm::Error("Stream error");
+ }
+ *rv = nbytes;
+ });
+
+ return (*event_handler_factor)(writer_func, name, remote_key);
+}
+
+/*!
+ * \brief Helper function to query real IP of device in WiFi network
+ * \return string with IPv4 in format "192.168.0.1" or "unknown" if cannot detect
+ */
+static std::string getWiFiAddress() {
+ std::string address = "unknown";
+ ifaddrs* interfaces = nullptr;
+
+ int success = getifaddrs(&interfaces);
+ if (success == 0) {
+ ifaddrs* temp_addr = interfaces;
+ while (temp_addr != NULL) {
+ if (temp_addr->ifa_addr->sa_family == AF_INET) {
+ // Check if interface is en0 which is the wifi connection on the iPhone
+ if (std::string(temp_addr->ifa_name) == "en0") {
+ address = inet_ntoa(((sockaddr_in*)temp_addr->ifa_addr)->sin_addr);
+ }
+ }
+ temp_addr = temp_addr->ifa_next;
+ }
+ }
+
+ freeifaddrs(interfaces);
+ return address;
+}
+
+} // namespace runtime
+} // namespace tvm
+
+// Base class for any type of RPC servicing
+@interface RPCServerBase : RPCServer
+
+/*!
+ * Methods should be implemented in inherited classes
+ */
+- (bool)onReadHandler; // return true - continue feeding, false - stop, try to drain output buffer
+- (bool)onWriteHandler; // return true - continue draining, false - no data to write
+- (void)onEndEncountered; // called on disconnect or session desided that it's shutdown time
+- (void)open; // Initiate listening objects like i/o streams and other resources
+- (void)close; // Deinitialize resources opend in "open" method
+@end
+
+@implementation RPCServerBase {
+ // Worker thread
+ NSThread* worker_thread_;
+ // Triger to continue RunLoop processing inside worker_thread_
+ BOOL shouldKeepRunning;
+ // Input socket stream
+ @protected
+ NSInputStream* inputStream_;
+ // Output socket stream
+ NSOutputStream* outputStream_;
+ // Temporal buffer with data to send
+ std::string sendBuffer_;
+ // Temporal receive buffer
+ std::string recvBuffer_;
+ // Requested data size to accumulate in recvBuffer_ before continue processing
+ int requiredToRecv_;
+}
+
+/*!
+ * Start internal worker thread with RunLoop and submit correspoding open handlers into it
+ * Not blocking
+ */
+- (void)start {
+ worker_thread_ = [[NSThread alloc] initWithBlock:^{
+ @autoreleasepool {
+ [self notifyState:RPCServerStatus_Launched];
+ [self open];
+ shouldKeepRunning = YES;
+ while (shouldKeepRunning && [[NSRunLoop currentRunLoop] runMode:NSDefaultRunLoopMode
+ beforeDate:[NSDate distantFuture]])
+ ;
+ [self notifyState:RPCServerStatus_Stopped];
+ }
+ }];
+ [worker_thread_ start];
+}
+
+/*!
+ * Send message to workel thread runloop to finish processing
+ * Not blocking
+ */
+- (void)stop {
+ if (worker_thread_ == nil) return;
+
+ [self performSelector:@selector(stop_) onThread:worker_thread_ withObject:nil waitUntilDone:NO];
+ worker_thread_ = nil; // TODO: is it valide? may be better to do that inside NSThread?
+}
+
+- (void)stop_ {
+ [self close];
+ shouldKeepRunning = NO;
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will connect to host and port specified in corresponding properties
+ */
+- (void)open {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)self.host, self.port, &readStream,
+ &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Base implementation to selup i/o streams
+ * Will assign i/o streams to provided socket connection.
+ */
+- (void)openWithSocket:(CFSocketNativeHandle)sock {
+ CFReadStreamRef readStream;
+ CFWriteStreamRef writeStream;
+ CFStreamCreatePairWithSocket(NULL, sock, &readStream, &writeStream);
+ inputStream_ = (__bridge NSInputStream*)readStream;
+ outputStream_ = (__bridge NSOutputStream*)writeStream;
+ [inputStream_ setDelegate:self];
+ [outputStream_ setDelegate:self];
+ [inputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ open];
+ [inputStream_ open];
+}
+
+/*!
+ * Close i/o streams assosiated with connection
+ */
+- (void)close {
+ [inputStream_ close];
+ [outputStream_ close];
+ [inputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [outputStream_ removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
+ [inputStream_ setDelegate:nil];
+ [outputStream_ setDelegate:nil];
+ inputStream_ = nil;
+ outputStream_ = nil;
+}
+
+/// Unimplemented stubs
+- (bool)onReadHandler {
+ return false;
+}
+- (bool)onWriteHandler {
+ return false;
+}
+- (void)onEndEncountered {
+}
+
+/*!
+ * Try to read data from stream and call processing hadnler
+ */
+- (void)tryToRead {
+ const int kBufferSize = 4 << 10; // 4kB buffer
+ const int prev_size = recvBuffer_.size();
+ recvBuffer_.resize(kBufferSize);
+ size_t nbytes = [inputStream_ read:(uint8_t*)recvBuffer_.data() + prev_size
+ maxLength:recvBuffer_.size() - prev_size];
+ recvBuffer_.resize(nbytes + prev_size);
+
+ // feed while it accept or requested particulat buffer size
+ while (!recvBuffer_.empty() && requiredToRecv_ <= recvBuffer_.size() && [self onReadHandler])
+ ;
+}
+
+/*!
+ * Try to write remaining data to stream and call processing hadnler
+ */
+- (void)tryToWrite {
+ if (!sendBuffer_.empty()) {
+ size_t nbytes = [outputStream_ write:(uint8_t*)sendBuffer_.data() maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, nbytes);
+ }
+ // call write handler while it want be called and space is available
+ while (sendBuffer_.empty() && [outputStream_ hasSpaceAvailable] && [self onWriteHandler])
+ ;
+}
+
+/*!
+ * Main event handler of socket stream events
+ */
+- (void)stream:(NSStream*)strm handleEvent:(NSStreamEvent)event {
+ std::string buffer;
+ switch (event) {
+ case NSStreamEventOpenCompleted: {
+ // Nothing
+ break;
+ }
+ case NSStreamEventHasBytesAvailable:
+ if (strm == inputStream_) {
+ [self tryToRead];
+ if ([outputStream_ hasSpaceAvailable]) [self tryToWrite];
+ }
+ break;
+ case NSStreamEventHasSpaceAvailable: {
+ if (strm == outputStream_) {
+ [self tryToWrite];
+ if ([inputStream_ hasBytesAvailable]) [self tryToRead];
+ }
+ break;
+ }
+ case NSStreamEventErrorOccurred: {
+ [self notifyError:[strm streamError].localizedDescription];
+ break;
+ }
+ case NSStreamEventEndEncountered: {
+ [self onEndEncountered];
+ break;
+ }
+ default: {
+ NSLog(@"Unknown event");
+ }
+ }
+}
+
+#pragma mark - Helpers
+
+/*!
+ * Set buffer to send into stream. Try to send immediatly or submit to lazy sending
+ * Non blocking operation
+ */
+- (void)toSend:(NSData*)data {
+ sendBuffer_.append(static_cast<const char*>(data.bytes), data.length);
+
+ // try to flush buffer
+ NSInteger sent_size = [outputStream_ write:(uint8_t*)sendBuffer_.data()
+ maxLength:sendBuffer_.size()];
+ sendBuffer_.erase(0, sent_size);
+}
+
+/*!
+ * Set buffer to send in packet format [size, data]. Behaviour is same as for toSend.
+ */
+- (void)toSendPacked:(NSData*)data {
+ int packet_size = data.length;
+ [self toSend:[NSData dataWithBytes:&packet_size length:sizeof(packet_size)]];
+ [self toSend:data];
+}
+
+/*!
+ */
+- (NSData*)requestInputDataWithSize:(NSInteger)size {
+ if (recvBuffer_.size() < size) {
+ requiredToRecv_ = size;
+ return nil;
+ }
+ NSData* res = [NSData dataWithBytes:recvBuffer_.data() length:size];
+ recvBuffer_.erase(0, size);
+ return res;
+}
+
+/*!
+ */
+- (NSData*)requestInputDataPacked {
+ int size;
+ if (recvBuffer_.size() < sizeof(size)) {
+ requiredToRecv_ = sizeof(size);
+ return nil;
+ }
+ size = *(int*)recvBuffer_.data();
+ if (recvBuffer_.size() < sizeof(size) + size) {
+ requiredToRecv_ = sizeof(size) + size;
+ return nil;
+ }
+ NSData* res = [NSData dataWithBytes:recvBuffer_.data() + sizeof(size) length:size];
+ recvBuffer_.erase(0, sizeof(size) + size);
+ return res;
+};
+
+#pragma mark - Notifiers
+
+/*!
+ * Notify external listener about error.
+ * Also print error message to std out in case of Verbose mode
+ */
+- (void)notifyError:(NSString*)msg {
+ // Duplicate error message in std output. Host launcher script may listen it.
+ if (self.verbose) NSLog(@"[IOS-RPC] ERROR: %@", msg);
+ if (self.delegate) [self.delegate onError:msg];
+}
+
+/*!
+ * Notify external listener about server state changes.
+ * Also print information to std out in case of Verbose mode
+ */
+- (void)notifyState:(RPCServerStatus)state {
+ // Duplicate sattus changing in std output. Host launcher script may listen it.
+ if (self.verbose) NSLog(@"[IOS-RPC] STATE: %d", state);
+ if (self.delegate != nil) [self.delegate onStatusChanged:state];
+}
+
+@end
+
+@interface RPCServerProxy : RPCServerBase
+@end
+
+typedef enum {
+ RPCServerProxyState_Idle,
+ RPCServerProxyState_HandshakeToSend,
+ RPCServerProxyState_HandshakeToRecv,
+ RPCServerProxyState_Processing,
+} RPCServerProxyState;
+
+@implementation RPCServerProxy {
+ /// Original TVM RPC event handler
+ tvm::runtime::FEventHandler handler_;
+ @protected
+ /// Sate of Proxy client implementation
+ RPCServerProxyState state_;
+}
+
+- (instancetype)init {
+ if (self = [super init]) {
+ handler_ = nullptr;
+ state_ = RPCServerProxyState_Idle;
+ }
+ return self;
+}
+
+/*!
+ * Implement matching of internat state on state available for outside users
+ */
+- (void)setState:(RPCServerProxyState)new_state {
+ // Send Connected notification because Proxy doesn't responce until client connected.
+ if (new_state == RPCServerProxyState_HandshakeToRecv)
+ [self notifyState:RPCServerStatus_Connected];
+ if (new_state == RPCServerProxyState_Idle) [self notifyState:RPCServerStatus_Disconnected];
+ if (state_ == RPCServerProxyState_HandshakeToRecv && new_state == RPCServerProxyState_Processing)
+ [self notifyState:RPCServerStatus_RPCSessionStarted];
+ if (state_ == RPCServerProxyState_Processing && new_state == RPCServerProxyState_Idle)
+ [self notifyState:RPCServerStatus_RPCSessionStarted];
+
+ state_ = new_state;
+}
+
+- (bool)onWriteHandler {
+ switch (state_) {
+ case RPCServerProxyState_HandshakeToSend: {
+ // Send together kRPCMagic and server descriptor because of Proxy
+ int code = tvm::runtime::kRPCMagic;
+ [self toSend:[NSData dataWithBytes:&code length:sizeof(code)]];
+
+ std::string full_key = std::string("server:") + self.key.UTF8String;
+ [self toSendPacked:[NSData dataWithBytes:full_key.data() length:full_key.size()]];
+
+ self.state = RPCServerProxyState_HandshakeToRecv;
+ return TRUE;
+ }
+ case RPCServerProxyState_Processing: {
+ try {
+ TVMByteArray dummy{nullptr, 0};
+ int flag = handler_(dummy, 2);
+ if (flag == 0) {
+ [self onEndEncountered];
+ }
+ return flag == 2;
+ } catch (const tvm::Error& e) {
+ [self close];
+ }
+ break;
+ }
+ default:
+ // Nothing
+ break;
+ }
+ return FALSE;
+}
+
+- (bool)onReadHandler {
+ switch (state_) {
+ case RPCServerProxyState_HandshakeToRecv: {
+ NSData* data = [self requestInputDataWithSize:sizeof(int)];
+ if (data == nil) return FALSE;
+
+ if (*(int*)data.bytes != tvm::runtime::kRPCMagic) {
+ [self notifyError:@"Wrong responce, is not RPC client."];
+ [self close];
+ return FALSE;
+ break;
+ }
+
+ handler_ = tvm::runtime::CreateServerEventHandler(outputStream_, "iphone", "%toinit");
+
+ self.state = RPCServerProxyState_Processing;
+ return TRUE;
+ break;
+ }
+ case RPCServerProxyState_Processing: {
+ int flag = 1;
+ if ([outputStream_ hasSpaceAvailable]) {
+ flag |= 2;
+ }
+ // always try to write
+ try {
+ TVMByteArray arr{recvBuffer_.data(), recvBuffer_.size()};
+ flag = handler_(arr, flag);
+ recvBuffer_.clear();
+ if (flag == 0) {
+ [self onEndEncountered];
+ }
+ return flag == 1;
+ } catch (const tvm::Error& e) {
+ [self close];
+ }
+ break;
+ }
+ default:
+ // Nothing
+ break;
+ }
+ return FALSE;
+}
+
+- (void)onEndEncountered {
+ // Automatic reconnection when session is finished.
Review comment:
worth logging?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org