[feat] defer decrypt, gen_key and encrypt

This commit put those gnupg operations to be run on external threads
limited by the amount of cores present on user machine.
Some gnupg calls spawn processes and communicating to them is a
synchronous operation, so running outside of a reactor should improve
response time by avoiding reactor locking.

1 files changed, 29 insertions, 11 deletions

diff --git a/src/leap/keymanager/openpgp.py b/src/leap/keymanager/openpgp.py
index d6481377..90640439 100644
--- a/src/leap/keymanager/openpgp.py
+++ b/src/leap/keymanager/openpgp.py
@@ -27,9 +27,11 @@ import io
 
 
 from datetime import datetime
+from multiprocessing import cpu_count
 from gnupg import GPG
 from gnupg.gnupg import GPGUtilities
 from twisted.internet import defer
+from twisted.internet.threads import deferToThread
 
 from leap.common.check import leap_assert, leap_assert_type, leap_check
 from leap.keymanager import errors
@@ -55,6 +57,16 @@ logger = logging.getLogger(__name__)
 # A temporary GPG keyring wrapped to provide OpenPGP functionality.
 #
 
+# This function will be used to call blocking GPG functions outside
+# of Twisted reactor and match the concurrent calls to the amount of CPU cores
+cpu_core_semaphore = defer.DeferredSemaphore(cpu_count())
+
+
+def from_thread(func, *args, **kwargs):
+    call = lambda: deferToThread(func, *args, **kwargs)
+    return cpu_core_semaphore.run(call)
+
+
 class TempGPGWrapper(object):
     """
     A context manager that wraps a temporary GPG keyring which only contains
@@ -253,6 +265,7 @@ class OpenPGPScheme(EncryptionScheme):
         # make sure the key does not already exist
         leap_assert(is_address(address), 'Not an user address: %s' % address)
 
+        @defer.inlineCallbacks
         def _gen_key(_):
             with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
                 # TODO: inspect result, or use decorator
@@ -264,7 +277,7 @@ class OpenPGPScheme(EncryptionScheme):
                     name_comment='')
                 logger.info("About to generate keys... "
                             "This might take SOME time.")
-                gpg.gen_key(params)
+                yield from_thread(gpg.gen_key, params)
                 logger.info("Keys for %s have been successfully "
                             "generated." % (address,))
                 pubkeys = gpg.list_keys()
@@ -293,7 +306,7 @@ class OpenPGPScheme(EncryptionScheme):
                         gpg.export_keys(key['fingerprint'], secret=secret))
                     d = self.put_key(openpgp_key, address)
                     deferreds.append(d)
-                return defer.gatherResults(deferreds)
+                yield defer.gatherResults(deferreds)
 
         def key_already_exists(_):
             raise errors.KeyAlreadyExists(address)
@@ -686,6 +699,7 @@ class OpenPGPScheme(EncryptionScheme):
             raise errors.GPGError(
                 'Failed to encrypt/decrypt: %s' % stderr)
 
+    @defer.inlineCallbacks
     def encrypt(self, data, pubkey, passphrase=None, sign=None,
                 cipher_algo='AES256'):
         """
@@ -700,8 +714,8 @@ class OpenPGPScheme(EncryptionScheme):
         :param cipher_algo: The cipher algorithm to use.
         :type cipher_algo: str
 
-        :return: The encrypted data.
-        :rtype: str
+        :return: A Deferred that will be fired with the encrypted data.
+        :rtype: defer.Deferred
 
         :raise EncryptError: Raised if failed encrypting for some reason.
         """
@@ -713,7 +727,8 @@ class OpenPGPScheme(EncryptionScheme):
             leap_assert(sign.private is True)
             keys.append(sign)
         with TempGPGWrapper(keys, self._gpgbinary) as gpg:
-            result = gpg.encrypt(
+            result = yield from_thread(
+                gpg.encrypt,
                 data, pubkey.fingerprint,
                 default_key=sign.key_id if sign else None,
                 passphrase=passphrase, symmetric=False,
@@ -724,11 +739,12 @@ class OpenPGPScheme(EncryptionScheme):
             # result.data  - (bool) contains the result of the operation
             try:
                 self._assert_gpg_result_ok(result)
-                return result.data
+                defer.returnValue(result.data)
             except errors.GPGError as e:
                 logger.error('Failed to decrypt: %s.' % str(e))
                 raise errors.EncryptError()
 
+    @defer.inlineCallbacks
     def decrypt(self, data, privkey, passphrase=None, verify=None):
         """
         Decrypt C{data} using private @{privkey} and verify with C{verify} key.
@@ -743,8 +759,9 @@ class OpenPGPScheme(EncryptionScheme):
         :param verify: The key used to verify a signature.
         :type verify: OpenPGPKey
 
-        :return: The decrypted data and if signature verifies
-        :rtype: (unicode, bool)
+        :return: Deferred that will fire with the decrypted data and
+                 if signature verifies (unicode, bool)
+        :rtype: Deferred
 
         :raise DecryptError: Raised if failed decrypting for some reason.
         """
@@ -756,8 +773,9 @@ class OpenPGPScheme(EncryptionScheme):
             keys.append(verify)
         with TempGPGWrapper(keys, self._gpgbinary) as gpg:
             try:
-                result = gpg.decrypt(
-                    data, passphrase=passphrase, always_trust=True)
+                result = yield from_thread(gpg.decrypt,
+                                           data, passphrase=passphrase,
+                                           always_trust=True)
                 self._assert_gpg_result_ok(result)
 
                 # verify signature
@@ -767,7 +785,7 @@ class OpenPGPScheme(EncryptionScheme):
                         verify.fingerprint == result.pubkey_fingerprint):
                     sign_valid = True
 
-                return (result.data, sign_valid)
+                defer.returnValue((result.data, sign_valid))
             except errors.GPGError as e:
                 logger.error('Failed to decrypt: %s.' % str(e))
                 raise errors.DecryptError(str(e))