decode: added ability to decoder to resuse device in multisession decode mode

Author: t-boiko

common/libs/VkCodecUtils/ProgramConfig.h

@@ -80,6 +80,9 @@ struct ProgramConfig {
         outputcrcPerFrame = false;
         outputcrc = false;
         crcOutputFile = nullptr;
+        numberOfDecodeWorkers = 0;
+        enableWorkerProcessesPoll = false;
+        ipcType = 0;
     }
 
     using ProgramArgs = std::vector<ArgSpec>;
@@ -187,8 +190,7 @@ struct ProgramConfig {
             {"--input", "-i", 1, "Input filename to decode",
                 [this](const char **args, const ProgramArgs &a) {
                     videoFileName = args[0];
-                    std::ifstream validVideoFileStream(videoFileName, std::ifstream::in);
-                    return (bool)validVideoFileStream;
+                    return true;
                 }},
             {"--output", "-o", 1, "Output filename to dump raw video to",
                 [this](const char **args, const ProgramArgs &a) {
@@ -322,6 +324,17 @@ struct ProgramConfig {
                     crcInitValue = crcInitValueTemp;
                     return true;
                 }},
+            {"--poll-of-processes", nullptr, 1, "Use poll of worker processes and specify number of workers.",
+                    [this](const char **args, const ProgramArgs &a) {
+                    enableWorkerProcessesPoll = true;
+                    numberOfDecodeWorkers = std::atoi(args[0]);
+                    return true;
+                }},
+            {"--files-to-decode", nullptr, 1, "Specify a file location where command lines for the poll of worker processes are saved.",
+                    [this](const char **args, const ProgramArgs &a) {
+                    fileListIpc = args[0];
+                    return true;
+                }},
         };
 
         for (int i = 1; i < argc; i++) {
@@ -391,6 +404,18 @@ struct ProgramConfig {
                 crcOutputFile = stdout;
             }
         }
+
+        if (!enableWorkerProcessesPoll) {
+            if (videoFileName.length() == 0) {
+                std::cerr << "Input file should be specified" << std::endl;
+                exit(EXIT_FAILURE);
+            }
+            std::ifstream validVideoFileStream(videoFileName, std::ifstream::in);
+            if (!(bool)validVideoFileStream) {
+                std::cerr << "Can't open input file: invalid file name" << std::endl;
+                exit(EXIT_FAILURE);
+            }
+        }
     }
 
     // Assuming we have the length as a parameter:
@@ -461,6 +486,7 @@ struct ProgramConfig {
     uint32_t decoderQueueSize;
     int32_t enablePostProcessFilter;
     uint32_t *crcOutput;
+    uint32_t numberOfDecodeWorkers;
     uint32_t enableStreamDemuxing : 1;
     uint32_t directMode : 1;
     uint32_t vsync : 1;
@@ -474,6 +500,9 @@ struct ProgramConfig {
     uint32_t outputy4m : 1;
     uint32_t outputcrc : 1;
     uint32_t outputcrcPerFrame : 1;
+    uint32_t enableWorkerProcessesPoll : 1;
+    uint32_t ipcType : 1;
+    std::string fileListIpc;
 };
 
 #endif /* _PROGRAMSETTINGS_H_ */

common/libs/VkCodecUtils/poll_manager.h

@@ -0,0 +1,77 @@
+#include <vector>
+#include <string>
+#include <iostream>
+#include <sstream>
+
+enum IPC_TYPE { UNIX_DOMAIN_SOCKETS = 0 };
+constexpr int DEFAULT_BUFLEN = 512;
+
+int usoc_manager(int isNoPresent, std::string& inputCmdsList);
+int clientConnectServer(std::string& recvbuf, const char* usocfilename = NULL);
+
+#ifdef _WIN32
+
+static int cloneTheProcess(int argc, const char** argv, PROCESS_INFORMATION& pi,  STARTUPINFO& si)
+{
+    ZeroMemory( &si, sizeof(si) );
+    si.cb = sizeof(si);
+    ZeroMemory( &pi, sizeof(pi) );
+    std::string argsToPass;
+    for (int i = 0; i < argc; i++) {
+        argsToPass += argv[i];
+        argsToPass += " ";
+    }
+    argsToPass += "spawn";
+    if( !CreateProcess( NULL,
+        (LPTSTR)argsToPass.c_str(),
+        NULL,
+        NULL,
+        FALSE,
+        0,
+        NULL,
+        NULL,
+        &si,
+        &pi)
+    )
+    {
+        printf( "CreateProcess failed (%d).\n", GetLastError() );
+        return -1;
+    }
+    return 0;
+}
+#endif
+
+static int parseCharArray(std::vector<std::string>& w, const char* messageString, int& argc, const char **argv) {
+    std::stringstream ss(messageString);
+    std::string word;
+    argc = 0;
+    std::cout << std::endl;
+    while (ss >> w[argc]) {
+        if (w[argc][0] == '~') {
+            w[argc] = getenv("HOME") + w[argc].substr(1);
+        }
+        if (w[argc].substr(0, 6) == "finish") {
+            printf("Received a request to finish this decode worker. The worker process is terminated (completed).\n");
+            return 0;
+        }
+        if (w[argc].substr(0, 6) == "nodata") {
+            printf("Received a request to wait for a data...\n");
+            return 0;
+        }
+        argv[argc] = w[argc].c_str();
+        argc++;
+    }
+    return argc >= 1;
+}
+
+static int receiveNewBitstream(IPC_TYPE ipcType, bool enableWorkerProcessesPoll, std::string& receivedMessage)
+{
+    if (!enableWorkerProcessesPoll) {
+        return 0;
+    }
+    int isDataReceived = 0;
+    if (ipcType == IPC_TYPE::UNIX_DOMAIN_SOCKETS) {
+        isDataReceived = clientConnectServer(receivedMessage);
+    }
+    return isDataReceived;
+}