<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def upload(bucket, aws_access_key, aws_secret_key, iterable, key, progress_cb=None, threads=5, replace=False, secure=True, connection=None):
 """ Upload data to s3 using the s3 multipart upload API. :param bucket: Name of the S3 bucket :type bucket: String :param aws_access_key: AWS access key id (optional) :type aws_access_key: String :param aws_secret_key: AWS access secret key (optional) :type aws_secret_key: String :param iterable: The data to upload. Each 'part' in the list. will be uploaded in parallel. Each part must be at least 5242880 bytes (5mb). :type iterable: An iterable object :param key: The name of the key (filename) to create in the s3 bucket :type key: String :param progress_cb: Progress callback, will be called with (part_no, uploaded, total) each time a progress update is available. (optional) :type progress_cb: function :param threads: the number of threads to use while uploading. (Default is 5) :type threads: int :param replace: will replace the key (filename) on S3 if set to true. (Default is false) :type replace: boolean :param secure: Use ssl when talking to s3. (Default is true) :type secure: boolean :param connection: Used for testing (optional) :type connection: S3 connection class :returns: void """


    if not connection:
        from boto.s3.connection import S3Connection as connection
        c = connection(aws_access_key, aws_secret_key, is_secure=secure)
    else:
        c = connection

    b = c.get_bucket(bucket)

    if not replace and b.lookup(key):
        raise Exception('s3 key ' + key + ' already exists')

    multipart_obj = b.initiate_multipart_upload(key)
    err_queue = queue.Queue()
    lock = threading.Lock()
    upload.counter = 0

    try:
        tpool = pool.ThreadPool(processes=threads)

        def check_errors():
            try:
                exc = err_queue.get(block=False)
            except queue.Empty:
                pass
            else:
                raise exc

        def waiter():
            while upload.counter >= threads:
                check_errors()
                time.sleep(0.1)

        def cb(err):
            if err:
                err_queue.put(err)
            with lock:
                upload.counter -= 1

        args = [multipart_obj.upload_part_from_file, progress_cb]

        for part_no, part in enumerate(iterable):
            part_no += 1
            tpool.apply_async(upload_part, args + [part_no, part], callback=cb)
            with lock:
                upload.counter += 1
            waiter()

        tpool.close()
        tpool.join()
        # Check for thread errors before completing the upload,
        # sometimes an error can be left unchecked until we
        # get to this point.
        check_errors()
        multipart_obj.complete_upload()
    except:
        multipart_obj.cancel_upload()
        tpool.terminate()
        raise