Revert "Refactor dispatch handling (#2452)"

Due to performance regression:
dc60fe9#commitcomment-33943711

This reverts commit dc60fe9.

cli/dispatch_minimal.rs

@@ -6,7 +6,6 @@
 //! message or a "minimal" message.
 use crate::state::ThreadSafeState;
 use deno::Buf;
-use deno::CoreOp;
 use deno::Op;
 use deno::PinnedBuf;
 use futures::Future;
@@ -18,16 +17,23 @@ const OP_WRITE: i32 = 2;
 #[derive(Copy, Clone, Debug, PartialEq)]
 // This corresponds to RecordMinimal on the TS side.
 pub struct Record {
+  pub promise_id: i32,
   pub op_id: i32,
   pub arg: i32,
   pub result: i32,
 }
 
 impl Into<Buf> for Record {
   fn into(self) -> Buf {
-    let vec = vec![DISPATCH_MINIMAL_TOKEN, self.op_id, self.arg, self.result];
+    let vec = vec![
+      DISPATCH_MINIMAL_TOKEN,
+      self.promise_id,
+      self.op_id,
+      self.arg,
+      self.result,
+    ];
     let buf32 = vec.into_boxed_slice();
-    let ptr = Box::into_raw(buf32) as *mut [u8; 4 * 4];
+    let ptr = Box::into_raw(buf32) as *mut [u8; 5 * 4];
     unsafe { Box::from_raw(ptr) }
   }
 }
@@ -39,32 +45,36 @@ pub fn parse_min_record(bytes: &[u8]) -> Option<Record> {
   let p = bytes.as_ptr();
   #[allow(clippy::cast_ptr_alignment)]
   let p32 = p as *const i32;
-  let s = unsafe { std::slice::from_raw_parts(p32, bytes.len() / 3) };
+  let s = unsafe { std::slice::from_raw_parts(p32, bytes.len() / 4) };
 
-  if s.len() < 4 {
+  if s.len() < 5 {
     return None;
   }
   let ptr = s.as_ptr();
-  let ints = unsafe { std::slice::from_raw_parts(ptr, 4) };
+  let ints = unsafe { std::slice::from_raw_parts(ptr, 5) };
   if ints[0] != DISPATCH_MINIMAL_TOKEN {
     return None;
   }
   Some(Record {
-    op_id: ints[1],
-    arg: ints[2],
-    result: ints[3],
+    promise_id: ints[1],
+    op_id: ints[2],
+    arg: ints[3],
+    result: ints[4],
   })
 }
 
 #[test]
 fn test_parse_min_record() {
-  let buf = vec![0xFE, 0xCA, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0];
+  let buf = vec![
+    0xFE, 0xCA, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0,
+  ];
   assert_eq!(
     parse_min_record(&buf),
     Some(Record {
-      op_id: 1,
-      arg: 2,
-      result: 3,
+      promise_id: 1,
+      op_id: 2,
+      arg: 3,
+      result: 4,
     })
   );
 
@@ -79,7 +89,8 @@ pub fn dispatch_minimal(
   state: &ThreadSafeState,
   mut record: Record,
   zero_copy: Option<PinnedBuf>,
-) -> CoreOp {
+) -> Op {
+  let is_sync = record.promise_id == 0;
   let min_op = match record.op_id {
     OP_READ => ops::read(record.arg, zero_copy),
     OP_WRITE => ops::write(record.arg, zero_copy),
@@ -104,7 +115,11 @@ pub fn dispatch_minimal(
     state.metrics_op_completed(buf.len());
     Ok(buf)
   }));
-  Op::Async(fut)
+  if is_sync {
+    Op::Sync(fut.wait().unwrap())
+  } else {
+    Op::Async(fut)
+  }
 }
 
 mod ops {

cli/errors.rs

@@ -243,20 +243,6 @@ pub fn no_buffer_specified() -> DenoError {
   new(ErrorKind::InvalidInput, String::from("no buffer specified"))
 }
 
-pub fn no_async_support() -> DenoError {
-  new(
-    ErrorKind::NoAsyncSupport,
-    String::from("op doesn't support async calls"),
-  )
-}
-
-pub fn no_sync_support() -> DenoError {
-  new(
-    ErrorKind::NoSyncSupport,
-    String::from("op doesn't support sync calls"),
-  )
-}
-
 #[derive(Debug)]
 pub enum RustOrJsError {
   Rust(DenoError),

cli/msg.fbs

@@ -136,8 +136,6 @@ enum ErrorKind: byte {
   OpNotAvaiable,
   WorkerInitFailed,
   UnixError,
-  NoAsyncSupport,
-  NoSyncSupport,
   ImportMapError,
 }
 
@@ -155,6 +153,7 @@ enum MediaType: byte {
 }
 
 table Base {
+  cmd_id: uint32;
   sync: bool = false;
   error_kind: ErrorKind = NoError;
   error: string;