concurrent encoding

pandora_client/__init__.py

@@ -638,7 +638,7 @@ class Client(object):
             post['files'] = files['files']
             post['volume'] = name
             print('sending list of files in %s (%s total)' % (name, len(post['files'])))
-            r = self.api.async('update', post)
+            r = self.api.later('update', post)
             # send empty list to get updated list of requested info/files/data
             post = {'info': {}}
             r = self.api.update(post)
@@ -770,7 +770,7 @@ class Client(object):
             print('sending info for %d files' % len(info))
             post = {'info': {}, 'upload': True}
             post['info'] = self.get_info_for_ids(info, prefix)
-            r = self.api.async('update', post)
+            r = self.api.later('update', post)
             # send empty list to get updated list of requested info/files/data
             post = {'info': {}}
             r = self.api.update(post)
@@ -989,6 +989,12 @@ class Client(object):
         server.run(self, args)
 
     def client(self, args):
+        threads = [t.split('=')[-1] for t in args if t.startswith('c=')]
+        if threads:
+            threads = int(threads[0])
+        else:
+            threads = 1
+        args = [a for a in args if not a.startswith('c=')]
         urls = [u for u in args if u.startswith('http:')]
         name = [u for u in args if u not in urls]
         if not name:
@@ -1016,7 +1022,7 @@ class Client(object):
         else:
             url = urls[0]
         from . import client
-        c = client.DistributedClient(url, name)
+        c = client.DistributedClient(url, name, threads)
         c.run()
 
 class API(ox.API):
@@ -1036,7 +1042,7 @@ class API(ox.API):
         if hasattr(self, 'taskStatus') and not hasattr(self, 'getTaskStatus'):
             self.getTaskStatus = self.taskStatus
 
-    def async(self, action, data, interval=5):
+    def later(self, action, data, interval=5):
         t = r = getattr(self, action)(data)
         if r['status']['code'] == 200:
             # wait for async task to finish

pandora_client/client.py

@@ -2,22 +2,46 @@
 # vi:si:et:sw=4:sts=4:ts=4
 from __future__ import division, with_statement, print_function, absolute_import
 
-import os
 import json
-import subprocess
-import time
+import os
 import socket
+import subprocess
 import sys
+import threading
+import time
 
 import requests
 
 from . import extract
 
+class Encoding:
+
+    def worker(self):
+        while True:
+            if not self.client.next():
+                return
+
+    def __init__(self, client, threads=1):
+        self.client = client
+        self.threads = []
+        for i in range(threads):
+            t = threading.Thread(target=self.worker)
+            t.start()
+            self.threads.append(t)
+
+    def join(self):
+        for t in self.threads:
+            t.join()
+
+
 class DistributedClient:
 
-    def __init__(self, url, name):
+    interrupted = False
+
+    def __init__(self, url, name, threads):
         self.url = url
         self.name = name
+        self.threads = threads
         self.supported_formats = extract.supported_formats()
 
     def ping(self, oshash):
@@ -40,8 +64,7 @@ class DistributedClient:
         url = '%s/next' % self.url
         data = requests.get(url).json()
         if 'oshash' in data:
-            self.encode(data['oshash'], data['cmd'], data['output'])
-            return True
+            return self.encode(data['oshash'], data['cmd'], data['output'])
         return False
 
     def encode(self, oshash, cmds, output):
@@ -49,6 +72,8 @@ class DistributedClient:
             cmds = [cmds]
         for cmd in cmds:
             cmd[0] = extract.command('ffmpeg')
+            if self.threads > 1:
+                cmd = cmd[:1] + ['-nostats', '-loglevel', 'error'] + cmd[1:]
             if 'webm' in cmd and not self.supported_formats['webm']:
                 print("ffmpeg is compiled without WebM support")
                 return
@@ -56,6 +81,8 @@ class DistributedClient:
                 print("ffmpeg is compiled without H.264 support")
                 return
             try:
+                if self.threads > 1:
+                    print('encode', oshash)
                 p = subprocess.Popen(cmd)
                 r = None
                 n = 0
@@ -75,17 +102,32 @@ class DistributedClient:
                 self.status(oshash, '')
                 if os.path.exists(output):
                     os.unlink(output)
+                if self.threads > 1:
+                    self.interrupted = True
+                    return False
                 sys.exit(1)
             if r != 0:
                 break
         if r == 0:
+            if self.threads > 1:
+                print('ok', oshash)
             self.upload(oshash, output)
         else:
+            if self.threads > 1:
+                print('error', oshash)
             self.status(oshash, 'failed')
         if os.path.exists(output):
             os.unlink(output)
+        return True
 
     def run(self):
+        if self.threads > 1:
+            enc = Encoding(self, self.threads)
+            enc.join()
+        else:
+            self.run_single()
+
+    def run_single(self):
         new = True
         while True:
             if not self.next():