Added user list sub system.

src/main.cpp

@@ -7,9 +7,11 @@
 #include <unistd.h>
 #include "conf.h"
 #include "crypto.h"
+#include "usr.h"
 #include "proc.h"
 
 using namespace std;
+using namespace shared;
 
 int parse_cmd(int argc, char **argv);
 
@@ -34,6 +36,15 @@ int main(int argc, char **argv)
         if (conf::ctx.command == "run")
         {
             //TODO
+            
+            usr::add_user("pku1");
+            usr::add_user("pku2");
+            usr::add_user("pku3");
+
+            proc::ContractExecArgs ctargs;
+            ctargs.users = &usr::users;
+
+            proc::exec_contract(ctargs);
         }
     }
 

src/proc.cpp

@@ -2,7 +2,6 @@
 #include <iostream>
 #include <stdlib.h>
 #include <unistd.h>
-#include <sys/ioctl.h>
 #include <sys/types.h>
 #include <sys/wait.h>
 #include <rapidjson/document.h>
@@ -12,6 +11,7 @@
 #include "conf.h"
 
 using namespace std;
+using namespace shared;
 
 namespace proc
 {
@@ -21,12 +21,11 @@ namespace proc
  */
 int contract_pid;
 
-void write_to_stdin(ContractExecArgs &msg);
+void write_to_stdin(ContractExecArgs &args);
 bool is_contract_running();
-int exec_contract(ContractExecArgs &msg);
-int read_contract_outputs(vector<ContractUser> users);
+int exec_contract(ContractExecArgs &args);
 
-int exec_contract(ContractExecArgs &msg)
+int exec_contract(ContractExecArgs &args)
 {
     if (is_contract_running())
     {
@@ -45,7 +44,7 @@ int exec_contract(ContractExecArgs &msg)
         chdir(conf::ctx.contractDir.data());
 
         //Write the contract args to the stdin (0) of the contract process.
-        write_to_stdin(msg);
+        write_to_stdin(args);
 
         char *args[] = {conf::cfg.binary.data(), conf::cfg.binargs.data(), NULL};
 
@@ -60,49 +59,29 @@ int exec_contract(ContractExecArgs &msg)
     return 1;
 }
 
-//Read per-user outputs produced by the contract process.
-int read_contract_outputs(vector<ContractUser> users)
-{
-    if (is_contract_running())
-    {
-        cerr << "Contract process still running.\n";
-        return 0;
-    }
-    
-    for (ContractUser user : users)
-    {
-        int fdout = user.outpipe[0];
-        int bytes_available = 0;
-        ioctl(fdout, FIONREAD, &bytes_available);
-
-        if (bytes_available > 0)
-        {
-            char data[bytes_available];
-            read(fdout, data, bytes_available);
-            cout << "user:" << user.pubkeyb64 << " reply: '" << data << "'" << endl;
-        }
-    }
-
-    return 1;
-}
-
-void write_to_stdin(ContractExecArgs &msg)
+/**
+ * Passes the input in the format:
+ * {
+ *   "version":"0.1",
+ *   "usrfd":{ "pk1":[fd0, fd1], "pk2":[fd0, fd1] }
+ * }
+ */ 
+void write_to_stdin(ContractExecArgs &args)
 {
     Document d;
     d.SetObject();
     Document::AllocatorType &allocator = d.GetAllocator();
     d.AddMember("version", StringRef(_HP_VERSION_), allocator);
 
-    Value users(kArrayType);
-    for (ContractUser user : msg.users)
+    Value users(kObjectType);
+    for (auto& [pk, user] : (*args.users))
     {
-        Value v;
-        v.SetObject();
-        v.AddMember("fdin", user.inpipe[0], allocator);
-        v.AddMember("fdout", user.outpipe[1], allocator);
-        users.PushBack(v, allocator);
+        Value fdlist(kArrayType);
+        fdlist.PushBack(user.inpipe[0], allocator);
+        fdlist.PushBack(user.outpipe[1], allocator);
+        users.AddMember(StringRef(user.pubkeyb64.data()), fdlist, allocator);
     }
-    d.AddMember("users", users, allocator);
+    d.AddMember("usrfd", users, allocator);
 
     StringBuffer buffer;
     Writer<StringBuffer> writer(buffer);

src/proc.h

@@ -2,27 +2,23 @@
 #define _HP_PROC_H_
 
 #include <cstdio>
-#include <vector>
+#include <map>
+#include "shared.h"
 
 using namespace std;
+using namespace shared;
 
 namespace proc
 {
 
-struct ContractUser
-{
-    string pubkeyb64;
-    int inpipe[2];  //from User to Contract
-    int outpipe[2]; //from Contract to User
-};
-
 struct ContractExecArgs
 {
-    vector<ContractUser> users;
+    map<string, ContractUser> *users;
+
+    //ContractExecArgs(map<string, ContractUser> &_users)
 };
 
-int exec_contract(ContractExecArgs &msg);
-int read_contract_outputs(vector<ContractUser> users);
+int exec_contract(ContractExecArgs &args);
 bool is_contract_running();
 
 } // namespace proc

src/shared.h

@@ -0,0 +1,31 @@
+#ifndef _HP_SHARED_H_
+#define _HP_SHARED_H_
+
+#include <string>
+#include <vector>
+
+using namespace std;
+
+namespace shared
+{
+
+struct ContractUser
+{
+    string pubkeyb64;
+    int inpipe[2];
+    int outpipe[2];
+    vector<unsigned char> outbuffer;
+
+    ContractUser(string _pubkeyb64, int _inpipe[2], int _outpipe[2])
+    {
+        pubkeyb64 = _pubkeyb64;
+        inpipe[0] = _inpipe[0];
+        inpipe[1] = _inpipe[1];
+        outpipe[0] = _outpipe[0];
+        outpipe[1] = _outpipe[1];
+    }
+};
+
+} // namespace usr
+
+#endif

src/usr.cpp

@@ -0,0 +1,83 @@
+#include <cstdio>
+#include <iostream>
+#include <unistd.h>
+#include <sys/ioctl.h>
+#include <sys/types.h>
+#include "proc.h"
+#include "usr.h"
+
+using namespace std;
+
+namespace usr
+{
+
+map<string, ContractUser> users;
+
+void add_user(string pubkeyb64)
+{
+    if (users.count(pubkeyb64) == 1)
+    {
+        cerr << pubkeyb64 << " already exist. Cannot add user.\n";
+        return;
+    }
+
+    int inpipe[2];
+    int outpipe[2];
+    if (pipe(inpipe) != 0 || pipe(outpipe) != 0) {
+        cerr << "User pipe creation failed. pubkey:" << pubkeyb64 << endl;
+        return;
+    }
+
+    users.insert(pair<string, ContractUser>(pubkeyb64, ContractUser(pubkeyb64, inpipe, outpipe)));
+}
+
+void remove_user(string pubkeyb64)
+{
+    if (users.count(pubkeyb64) == 0)
+    {
+        cerr << pubkeyb64 << " does not exist. Cannot remove user.\n";
+        return;
+    }
+
+    auto itr = users.find(pubkeyb64);
+    ContractUser user = itr->second;
+    close(user.inpipe[0]);
+    close(user.inpipe[1]);
+    close(user.outpipe[0]);
+    close(user.outpipe[1]);
+
+    users.erase(itr);
+}
+
+//Read per-user outputs produced by the contract process.
+int read_contract_user_outputs()
+{
+    if (proc::is_contract_running())
+    {
+        cerr << "Contract process still running.\n";
+        return 0;
+    }
+
+    for (auto& [pk, user] : users)
+    {
+        int fdout = user.outpipe[0];
+        int bytes_available = 0;
+        ioctl(fdout, FIONREAD, &bytes_available);
+
+        if (bytes_available > 0)
+        {
+            unsigned char data[bytes_available];
+            read(fdout, data, bytes_available);
+
+            //Replace the existing user buffer with new buffer
+            vector<unsigned char> buffer(data, data + bytes_available);
+            user.outbuffer.swap(buffer);
+
+            cout << "Read " + to_string(bytes_available) << " bytes into user output buffer. user:" + user.pubkeyb64 << endl;
+        }
+    }
+
+    return 1;
+}
+
+} // namespace usr

src/usr.h

@@ -0,0 +1,20 @@
+#ifndef _HP_USR_H_
+#define _HP_USR_H_
+
+#include <cstdio>
+#include <vector>
+#include <map>
+#include "shared.h"
+
+using namespace std;
+using namespace shared;
+
+namespace usr
+{
+extern map<string, ContractUser> users;
+void add_user(string pubkeyb64);
+void remove_user(string pubkeyb64);
+
+} // namespace usr
+
+#endif