def multiget(client, keys, **options):
    """Executes a parallel-fetch across multiple threads. Returns a list
    containing :class:`~riak.riak_object.RiakObject` or
    :class:`~riak.datatypes.Datatype` instances, or 4-tuples of
    bucket-type, bucket, key, and the exception raised.

    If a ``pool`` option is included, the request will use the given worker
    pool and not a transient :class:`~riak.client.multi.MultiGetPool`. This
    option will be passed by the client if the ``multiget_pool_size``
    option was set on client initialization.

    :param client: the client to use
    :type client: :class:`~riak.client.RiakClient`
    :param keys: the keys to fetch in parallel
    :type keys: list of three-tuples -- bucket_type/bucket/key
    :param options: request options to
        :meth:`RiakBucket.get <riak.bucket.RiakBucket.get>`
    :type options: dict
    :rtype: list

    """
    transient_pool = False
    outq = Queue()

    if 'pool' in options:
        pool = options['pool']
        del options['pool']
    else:
        pool = MultiGetPool()
        transient_pool = True

    try:
        pool.start()
        for bucket_type, bucket, key in keys:
            task = Task(client, outq, bucket_type, bucket, key, None, options)
            pool.enq(task)

        results = []
        for _ in range(len(keys)):
            if pool.stopped():
                raise RuntimeError(
                        'Multi-get operation interrupted by pool '
                        'stopping!')
            results.append(outq.get())
            outq.task_done()
    finally:
        if transient_pool:
            pool.stop()

    return results