WebSocket Pepper API: synchronous completion support.

Allow optional remote callback, then perform synchronous completion
if receiving data is enough small for SRPC buffer. Otherwise invoke
another callback for asynchronous completion from main thread.

BUG=87310
TEST=browser_tests --gtest_filter=PPAPINaclTest.WebSocket_StressedSendReceive

Review URL: https://chromiumcodereview.appspot.com/9802027

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@132413 0039d316-1c4b-4281-b951-d872f2087c98

chrome/test/ui/ppapi_uitest.cc

@@ -886,6 +886,7 @@ TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_TextSendReceive)
 TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_BinarySendReceive)
 TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_StressedSendReceive)
 TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_BufferedAmount)
+TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_AbortCalls)
 TEST_PPAPI_IN_PROCESS_WITH_WS(WebSocket_CcInterfaces)
 TEST_PPAPI_IN_PROCESS(WebSocket_UtilityInvalidConnect)
 TEST_PPAPI_IN_PROCESS(WebSocket_UtilityProtocols)
@@ -910,6 +911,7 @@ TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_TextSendReceive)
 TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_BinarySendReceive)
 TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_StressedSendReceive)
 TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_BufferedAmount)
+TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_AbortCalls)
 TEST_PPAPI_NACL_VIA_HTTP_WITH_WS(WebSocket_CcInterfaces)
 TEST_PPAPI_NACL_VIA_HTTP(WebSocket_UtilityInvalidConnect)
 TEST_PPAPI_NACL_VIA_HTTP(WebSocket_UtilityProtocols)

ppapi/native_client/src/shared/ppapi_proxy/browser_ppb_websocket_rpc_server.cc

@@ -11,14 +11,17 @@
 #include "native_client/src/shared/ppapi_proxy/utility.h"
 #include "ppapi/c/pp_completion_callback.h"
 #include "ppapi/c/pp_errors.h"
-#include "srpcgen/ppb_rpc.h"
+#include "ppapi/c/ppb_core.h"
 #include "ppapi/c/ppb_websocket.h"
+#include "srpcgen/ppb_rpc.h"
 
 using ppapi_proxy::DebugPrintf;
 using ppapi_proxy::DeleteRemoteCallbackInfo;
 using ppapi_proxy::DeserializeTo;
 using ppapi_proxy::MakeRemoteCompletionCallback;
+using ppapi_proxy::PPBCoreInterface;
 using ppapi_proxy::PPBWebSocketInterface;
+using ppapi_proxy::Serialize;
 using ppapi_proxy::SerializeTo;
 
 void PpbWebSocketRpcServer::PPB_WebSocket_Create(
@@ -135,7 +138,9 @@ void PpbWebSocketRpcServer::PPB_WebSocket_ReceiveMessage(
     PP_Resource ws,
     int32_t callback_id,
     // outputs
-    int32_t* pp_error) {
+    int32_t* pp_error,
+    nacl_abi_size_t* sync_read_buffer_size,
+    char* sync_read_buffer_bytes) {
   NaClSrpcClosureRunner runner(done);
   rpc->result = NACL_SRPC_RESULT_APP_ERROR;
 
@@ -144,22 +149,31 @@ void PpbWebSocketRpcServer::PPB_WebSocket_ReceiveMessage(
       MakeRemoteCompletionCallback(rpc->channel, callback_id, &callback_var);
   if (NULL == remote_callback.func)
     return;
-  // TODO(toyoshim): Removing optional flag is easy way to expect asynchronous
-  // completion on the following PPBWebSocketInterface()->ReceiveMessage(). But
-  // from the viewpoint of performance, we should handle synchronous
-  // completion correctly.
-  remote_callback.flags &= ~PP_COMPLETIONCALLBACK_FLAG_OPTIONAL;
-
-  // The callback is always invoked asynchronously for now, so it doesn't care
-  // about re-entrancy.
+
   *pp_error = PPBWebSocketInterface()->ReceiveMessage(
       ws, callback_var, remote_callback);
   DebugPrintf("PPB_WebSocket::ReceiveMessage: pp_error=%"NACL_PRId32"\n",
       *pp_error);
-  CHECK(*pp_error != PP_OK);  // Should not complete synchronously
-  if (*pp_error != PP_OK_COMPLETIONPENDING)
-    DeleteRemoteCallbackInfo(remote_callback);
   rpc->result = NACL_SRPC_RESULT_OK;
+
+  // No callback scheduled. Handles synchronous completion here.
+  if (*pp_error != PP_OK_COMPLETIONPENDING) {
+    if (*pp_error == PP_OK) {
+      // Try serialization from callback_var to sync_read_buffer_bytes. It
+      // could fail if serialized callback_var is larger than
+      // sync_read_buffer_size.
+      if (!SerializeTo(callback_var, sync_read_buffer_bytes,
+          sync_read_buffer_size)) {
+        // Buffer for synchronous completion is not big enough. Uses
+        // asynchronous completion callback.
+        *pp_error = PP_OK_COMPLETIONPENDING;
+        // Schedule to invoke remote_callback later from main thread.
+        PPBCoreInterface()->CallOnMainThread(0, remote_callback, PP_OK);
+        return;
+      }
+    }
+    DeleteRemoteCallbackInfo(remote_callback);
+  }
 }
 
 void PpbWebSocketRpcServer::PPB_WebSocket_SendMessage(

ppapi/native_client/src/shared/ppapi_proxy/plugin_ppb_websocket.cc

@@ -120,25 +120,50 @@ int32_t ReceiveMessage(PP_Resource ws,
                        PP_CompletionCallback callback) {
   DebugPrintf("PPB_WebSocket::ReceiveMessage: ws=%"NACL_PRId32"\n", ws);
 
+  if (message == NULL)
+    return PP_ERROR_FAILED;
   int32_t callback_id =
       CompletionCallbackTable::Get()->AddCallback(callback, message);
   if (callback_id == 0)
     return PP_ERROR_BLOCKS_MAIN_THREAD;
 
-  // TODO(toyoshim): ReceiveMessage needs performance optimization to reduce
-  // chances to call RPC.
+  nacl_abi_size_t sync_read_buffer_size = kMaxReturnVarSize;
+  nacl::scoped_array<char> sync_read_buffer_bytes(
+      new char[sync_read_buffer_size]);
+
   int32_t pp_error;
   NaClSrpcError srpc_result =
       PpbWebSocketRpcClient::PPB_WebSocket_ReceiveMessage(
           GetMainSrpcChannel(),
           ws,
           callback_id,
-          &pp_error);
+          &pp_error,
+          &sync_read_buffer_size,
+          sync_read_buffer_bytes.get());
   DebugPrintf("PPB_WebSocket::ReceiveMessage: %s\n",
       NaClSrpcErrorString(srpc_result));
 
   if (srpc_result != NACL_SRPC_RESULT_OK)
     pp_error = PP_ERROR_FAILED;
+
+  if (pp_error != PP_OK_COMPLETIONPENDING) {
+    // Consumes plugin callback and deserialize received data.
+    void* plugin_buffer;
+    PP_Var* plugin_var;
+    PP_CompletionCallback plugin_callback =
+        CompletionCallbackTable::Get()->RemoveCallback(callback_id,
+                                                       &plugin_buffer,
+                                                       &plugin_var);
+    DCHECK(plugin_callback.func == callback.func);
+    DCHECK(plugin_var == message);
+    if (pp_error == PP_OK) {
+      *message = PP_MakeUndefined();
+      if (!DeserializeTo(sync_read_buffer_bytes.get(),
+          sync_read_buffer_size, 1, message))
+        return PP_ERROR_FAILED;
+    }
+  }
+
   return MayForceCallback(callback, pp_error);
 }
 

ppapi/native_client/src/shared/ppapi_proxy/ppb_rpc_client.cc

@@ -3574,17 +3574,19 @@ NaClSrpcError PpbWebSocketRpcClient::PPB_WebSocket_ReceiveMessage(
     NaClSrpcChannel* channel,
     PP_Resource ws,
     int32_t callback_id,
-    int32_t* pp_error)  {
+    int32_t* pp_error,
+    nacl_abi_size_t* sync_read_buffer_bytes, char* sync_read_buffer)  {
   VCHECK(ppapi_proxy::PPBCoreInterface()->IsMainThread(),
          ("%s: PPAPI calls are not supported off the main thread\n",
           __FUNCTION__));
   NaClSrpcError retval;
   retval = NaClSrpcInvokeBySignature(
       channel,
-      "PPB_WebSocket_ReceiveMessage:ii:i",
+      "PPB_WebSocket_ReceiveMessage:ii:iC",
       ws,
       callback_id,
-      pp_error
+      pp_error,
+      sync_read_buffer_bytes, sync_read_buffer
   );
   return retval;
 }

ppapi/native_client/src/shared/ppapi_proxy/ppb_rpc_server.cc

@@ -2871,7 +2871,8 @@ static void PPB_WebSocket_ReceiveMessageDispatcher(
       done,
       inputs[0]->u.ival,
       inputs[1]->u.ival,
-      &(outputs[0]->u.ival)
+      &(outputs[0]->u.ival),
+      &(outputs[1]->u.count), outputs[1]->arrays.carr
   );
 }
 
@@ -3297,7 +3298,7 @@ NaClSrpcHandlerDesc PpbRpcs::srpc_methods[] = {
   { "PPB_WebSocket_IsWebSocket:i:i", PPB_WebSocket_IsWebSocketDispatcher },
   { "PPB_WebSocket_Connect:iCCii:i", PPB_WebSocket_ConnectDispatcher },
   { "PPB_WebSocket_Close:iiCi:i", PPB_WebSocket_CloseDispatcher },
-  { "PPB_WebSocket_ReceiveMessage:ii:i", PPB_WebSocket_ReceiveMessageDispatcher },
+  { "PPB_WebSocket_ReceiveMessage:ii:iC", PPB_WebSocket_ReceiveMessageDispatcher },
   { "PPB_WebSocket_SendMessage:iC:i", PPB_WebSocket_SendMessageDispatcher },
   { "PPB_WebSocket_GetBufferedAmount:i:l", PPB_WebSocket_GetBufferedAmountDispatcher },
   { "PPB_WebSocket_GetCloseCode:i:i", PPB_WebSocket_GetCloseCodeDispatcher },

ppapi/native_client/src/shared/ppapi_proxy/ppb_websocket.srpc

@@ -44,6 +44,7 @@
                       ['callback_id', 'int32_t'],      # PP_CompletionCallback
                      ],
            'outputs': [['pp_error', 'int32_t'],        # int32_t
+                       ['sync_read_buffer', 'char[]'], # PP_Var
                       ]
           },
           {'name': 'PPB_WebSocket_SendMessage',

ppapi/native_client/src/shared/ppapi_proxy/trusted/srpcgen/ppb_rpc.h

@@ -1430,7 +1430,8 @@ class PpbWebSocketRpcServer {
       NaClSrpcClosure* done,
       PP_Resource ws,
       int32_t callback_id,
-      int32_t* pp_error);
+      int32_t* pp_error,
+      nacl_abi_size_t* sync_read_buffer_bytes, char* sync_read_buffer);
   static void PPB_WebSocket_SendMessage(
       NaClSrpcRpc* rpc,
       NaClSrpcClosure* done,

ppapi/native_client/src/shared/ppapi_proxy/untrusted/srpcgen/ppb_rpc.h

@@ -1243,7 +1243,8 @@ class PpbWebSocketRpcClient {
       NaClSrpcChannel* channel,
       PP_Resource ws,
       int32_t callback_id,
-      int32_t* pp_error);
+      int32_t* pp_error,
+      nacl_abi_size_t* sync_read_buffer_bytes, char* sync_read_buffer);
   static NaClSrpcError PPB_WebSocket_SendMessage(
       NaClSrpcChannel* channel,
       PP_Resource ws,

ppapi/tests/test_websocket.cc

@@ -6,7 +6,9 @@
 
 #include <stdio.h>
 #include <string.h>
+
 #include <algorithm>
+#include <memory>
 #include <string>
 #include <vector>
 
@@ -198,6 +200,7 @@ void TestWebSocket::RunTests(const std::string& filter) {
   RUN_TEST_WITH_REFERENCE_CHECK(BinarySendReceive, filter);
   RUN_TEST_WITH_REFERENCE_CHECK(StressedSendReceive, filter);
   RUN_TEST_WITH_REFERENCE_CHECK(BufferedAmount, filter);
+  RUN_TEST_WITH_REFERENCE_CHECK(AbortCalls, filter);
 
   RUN_TEST_WITH_REFERENCE_CHECK(CcInterfaces, filter);
 
@@ -649,28 +652,43 @@ std::string TestWebSocket::TestStressedSendReceive() {
   for (uint32_t i = 0; i < binary.size(); ++i)
     binary[i] = i;
   PP_Var binary_var = CreateVarBinary(binary);
+  // Prepare very large binary data over 64KiB. Object serializer in
+  // ppapi_proxy has a limitation of 64KiB as maximum return PP_Var data size
+  // to SRPC. In case received data over 64KiB exists, a specific code handles
+  // this large data via asynchronous callback from main thread. This data
+  // intends to test the code.
+  std::vector<uint8_t> large_binary(65 * 1024);
+  for (uint32_t i = 0; i < large_binary.size(); ++i)
+    large_binary[i] = i & 0xff;
+  PP_Var large_binary_var = CreateVarBinary(large_binary);
 
   // Send many messages.
+  int32_t result;
   for (int i = 0; i < 256; ++i) {
-    int32_t result = websocket_interface_->SendMessage(ws, text_var);
+    result = websocket_interface_->SendMessage(ws, text_var);
     ASSERT_EQ(PP_OK, result);
     result = websocket_interface_->SendMessage(ws, binary_var);
     ASSERT_EQ(PP_OK, result);
   }
+  result = websocket_interface_->SendMessage(ws, large_binary_var);
+  ASSERT_EQ(PP_OK, result);
   ReleaseVar(text_var);
   ReleaseVar(binary_var);
+  ReleaseVar(large_binary_var);
 
   // Receive echoed data.
-  for (int i = 0; i < 512; ++i) {
+  for (int i = 0; i <= 512; ++i) {
     TestCompletionCallback callback(instance_->pp_instance(), force_async_);
     PP_Var received_message;
-    int32_t result = websocket_interface_->ReceiveMessage(
+    result = websocket_interface_->ReceiveMessage(
         ws, &received_message, callback.GetCallback().pp_completion_callback());
     ASSERT_TRUE(result == PP_OK || result == PP_OK_COMPLETIONPENDING);
     if (result == PP_OK_COMPLETIONPENDING)
       result = callback.WaitForResult();
     ASSERT_EQ(PP_OK, result);
-    if (i & 1) {
+    if (i == 512) {
+      ASSERT_TRUE(AreEqualWithBinary(received_message, large_binary));
+    } else if (i & 1) {
       ASSERT_TRUE(AreEqualWithBinary(received_message, binary));
     } else {
       ASSERT_TRUE(AreEqualWithString(received_message, text));
@@ -746,6 +764,147 @@ std::string TestWebSocket::TestBufferedAmount() {
   PASS();
 }
 
+std::string TestWebSocket::TestAbortCalls() {
+  // Test abort behaviors where a WebSocket PP_Resource is released while
+  // each function is in-flight on the WebSocket PP_Resource.
+  std::vector<uint8_t> large_binary(65 * 1024);
+  PP_Var large_var = CreateVarBinary(large_binary);
+
+  // Firstly, test the behavior for SendMessage().
+  // This function doesn't require a callback, but operation will be done
+  // asynchronously in WebKit and browser process.
+  int32_t result;
+  std::string url = GetFullURL(kEchoServerURL);
+  PP_Resource ws = Connect(url, &result, "");
+  ASSERT_TRUE(ws);
+  ASSERT_EQ(PP_OK, result);
+  result = websocket_interface_->SendMessage(ws, large_var);
+  ASSERT_EQ(PP_OK, result);
+  core_interface_->ReleaseResource(ws);
+
+  // Following tests make sure the behavior for functions which require a
+  // callback. The callback must get a PP_ERROR_ABORTED.
+  // Test the behavior for Connect().
+  ws = websocket_interface_->Create(instance_->pp_instance());
+  ASSERT_TRUE(ws);
+  PP_Var url_var = CreateVarString(url);
+  TestCompletionCallback connect_callback(
+      instance_->pp_instance(), force_async_);
+  result = websocket_interface_->Connect(ws, url_var, NULL, 0,
+      connect_callback.GetCallback().pp_completion_callback());
+  ASSERT_EQ(PP_OK_COMPLETIONPENDING, result);
+  core_interface_->ReleaseResource(ws);
+  result = connect_callback.WaitForResult();
+  ASSERT_EQ(PP_ERROR_ABORTED, result);
+  ReleaseVar(url_var);
+
+  // Test the behavior for Close().
+  ws = Connect(url, &result, "");
+  ASSERT_TRUE(ws);
+  ASSERT_EQ(PP_OK, result);
+  PP_Var reason_var = CreateVarString("abort");
+  TestCompletionCallback close_callback(
+      instance_->pp_instance(), force_async_);
+  result = websocket_interface_->Close(ws,
+      PP_WEBSOCKETSTATUSCODE_NORMAL_CLOSURE, reason_var,
+      close_callback.GetCallback().pp_completion_callback());
+  ASSERT_EQ(PP_OK_COMPLETIONPENDING, result);
+  core_interface_->ReleaseResource(ws);
+  result = close_callback.WaitForResult();
+  ASSERT_EQ(PP_ERROR_ABORTED, result);
+  ReleaseVar(reason_var);
+
+  // Test the behavior for ReceiveMessage().
+  // Firstly, make sure the simplest case to wait for data which never arrives.
+  ws = Connect(url, &result, "");
+  ASSERT_TRUE(ws);
+  ASSERT_EQ(PP_OK, result);
+  PP_Var receive_var;
+  TestCompletionCallback receive_callback(
+      instance_->pp_instance(), force_async_);
+  result = websocket_interface_->ReceiveMessage(ws, &receive_var,
+      receive_callback.GetCallback().pp_completion_callback());
+  ASSERT_EQ(PP_OK_COMPLETIONPENDING, result);
+  core_interface_->ReleaseResource(ws);
+  result = receive_callback.WaitForResult();
+  ASSERT_EQ(PP_ERROR_ABORTED, result);
+
+  // Test the behavior where receive process might be in-flight.
+  const char* text = "yukarin";
+  PP_Var text_var = CreateVarString(text);
+
+  // Each trial sends 17 messages and receives just |trial| number of
+  // message(s) before releasing the WebSocket. The WebSocket is released while
+  // the next message is going to be received.
+  for (int trial = 1; trial <= 16; trial++) {
+    ws = Connect(url, &result, "");
+    ASSERT_TRUE(ws);
+    ASSERT_EQ(PP_OK, result);
+    for (int i = 0; i <= 16; ++i) {
+      result = websocket_interface_->SendMessage(ws, text_var);
+      ASSERT_EQ(PP_OK, result);
+    }
+    std::auto_ptr<TestCompletionCallback> callback;
+    PP_Var var;
+    for (int i = 0; i < trial; ++i) {
+      callback.reset(
+          new TestCompletionCallback(instance_->pp_instance(), force_async_));
+      result = websocket_interface_->ReceiveMessage(
+          ws, &var, callback->GetCallback().pp_completion_callback());
+      if (result == PP_OK_COMPLETIONPENDING)
+        result = callback->WaitForResult();
+      ASSERT_EQ(PP_OK, result);
+      ASSERT_TRUE(AreEqualWithString(var, text));
+      ReleaseVar(var);
+    }
+    result = websocket_interface_->ReceiveMessage(
+        ws, &var, callback->GetCallback().pp_completion_callback());
+    core_interface_->ReleaseResource(ws);
+    if (result != PP_OK) {
+      result = callback->WaitForResult();
+      ASSERT_EQ(PP_ERROR_ABORTED, result);
+    }
+  }
+  // Same test, but the last receiving message is large message over 64KiB.
+  for (int trial = 1; trial <= 16; trial++) {
+    ws = Connect(url, &result, "");
+    ASSERT_TRUE(ws);
+    ASSERT_EQ(PP_OK, result);
+    for (int i = 0; i <= 16; ++i) {
+      if (i == trial)
+        result = websocket_interface_->SendMessage(ws, large_var);
+      else
+        result = websocket_interface_->SendMessage(ws, text_var);
+      ASSERT_EQ(PP_OK, result);
+    }
+    std::auto_ptr<TestCompletionCallback> callback;
+    PP_Var var;
+    for (int i = 0; i < trial; ++i) {
+      callback.reset(
+          new TestCompletionCallback(instance_->pp_instance(), force_async_));
+      result = websocket_interface_->ReceiveMessage(
+          ws, &var, callback->GetCallback().pp_completion_callback());
+      if (result == PP_OK_COMPLETIONPENDING)
+        result = callback->WaitForResult();
+      ASSERT_EQ(PP_OK, result);
+      ASSERT_TRUE(AreEqualWithString(var, text));
+      ReleaseVar(var);
+    }
+    result = websocket_interface_->ReceiveMessage(
+        ws, &var, callback->GetCallback().pp_completion_callback());
+    core_interface_->ReleaseResource(ws);
+    if (result != PP_OK) {
+      result = callback->WaitForResult();
+      ASSERT_EQ(PP_ERROR_ABORTED, result);
+    }
+  }
+
+  ReleaseVar(large_var);
+  ReleaseVar(text_var);
+
+  PASS();
+}
+
 std::string TestWebSocket::TestCcInterfaces() {
   // C++ bindings is simple straightforward, then just verifies interfaces work
   // as a interface bridge fine.
@@ -1152,3 +1311,4 @@ std::string TestWebSocket::TestUtilityBufferedAmount() {
 
   PASS();
 }
+

ppapi/tests/test_websocket.h

@@ -47,6 +47,7 @@ class TestWebSocket : public TestCase {
   std::string TestBinarySendReceive();
   std::string TestStressedSendReceive();
   std::string TestBufferedAmount();
+  std::string TestAbortCalls();
 
   std::string TestCcInterfaces();