S3Transfer.upload_file calls callback twice
As per the [examples in the docs](https://boto3.readthedocs.org/en/latest/reference/customizations/s3.html#usage), passing a callback to `S3Transfer.upload_file` displays the upload progress. However because `botocore.auth.SigV4Auth.payload` consumes the request body this actually happens twice, once for calculating the SHA256 of the body and one more when sending the body. This is somewhat cumbersome to workaround, I did it by adding these lines to the `Progress.__call__` example:

```
            if bytes_amount > 0 and self._seen_so_far == self._size:
                # assume we are now only starting the actual transfer
                self._seen_so_far = 0
```

It would be nice if somehow this could be done a bit better, seems however that botocore has no support for progress reporting and I don't see an obvious way to do it correctly (apart from the present solution).

diff --git a/boto3/s3/transfer.py b/boto3/s3/transfer.py
index aa4eb1ac..3258d8a9 100644
--- a/boto3/s3/transfer.py
+++ b/boto3/s3/transfer.py
@@ -152,13 +152,25 @@ def random_file_extension(num_digits=8):
     return ''.join(random.choice(string.hexdigits) for _ in range(num_digits))
 
 
+def disable_upload_callbacks(request, operation_name, **kwargs):
+    if operation_name in ['PutObject', 'UploadPart'] and \\
+            hasattr(request.body, 'disable_callback'):
+        request.body.disable_callback()
+
+
+def enable_upload_callbacks(request, operation_name, **kwargs):
+    if operation_name in ['PutObject', 'UploadPart'] and \\
+            hasattr(request.body, 'enable_callback'):
+        request.body.enable_callback()
+
+
 class QueueShutdownError(Exception):
     pass
 
 
 class ReadFileChunk(object):
     def __init__(self, fileobj, start_byte, chunk_size, full_file_size,
-                 callback=None):
+                 callback=None, enable_callback=True):
         """
 
         Given a file object shown below:
@@ -195,9 +207,11 @@ class ReadFileChunk(object):
         self._fileobj.seek(self._start_byte)
         self._amount_read = 0
         self._callback = callback
+        self._callback_enabled = enable_callback
 
     @classmethod
-    def from_filename(cls, filename, start_byte, chunk_size, callback=None):
+    def from_filename(cls, filename, start_byte, chunk_size, callback=None,
+                      enable_callback=True):
         """Convenience factory function to create from a filename.
 
         :type start_byte: int
@@ -215,13 +229,18 @@ class ReadFileChunk(object):
         :type callback: function(amount_read)
         :param callback: Called whenever data is read from this object.
 
+        :type enable_callback: bool
+        :param enable_callback: Indicate whether to invoke callback
+            during read() calls.
+
         :rtype: ``ReadFileChunk``
         :return: A new instance of ``ReadFileChunk``
 
         """
         f = open(filename, 'rb')
         file_size = os.fstat(f.fileno()).st_size
-        return cls(f, start_byte, chunk_size, file_size, callback)
+        return cls(f, start_byte, chunk_size, file_size, callback,
+                   enable_callback)
 
     def _calculate_file_size(self, fileobj, requested_size, start_byte,
                              actual_file_size):
@@ -235,10 +254,16 @@ class ReadFileChunk(object):
             amount_to_read = min(self._size - self._amount_read, amount)
         data = self._fileobj.read(amount_to_read)
         self._amount_read += len(data)
-        if self._callback is not None:
+        if self._callback is not None and self._callback_enabled:
             self._callback(len(data))
         return data
 
+    def enable_callback(self):
+        self._callback_enabled = True
+
+    def disable_callback(self):
+        self._callback_enabled = False
+
     def seek(self, where):
         self._fileobj.seek(self._start_byte + where)
         self._amount_read = where
@@ -291,7 +316,8 @@ class OSUtils(object):
 
     def open_file_chunk_reader(self, filename, start_byte, size, callback):
         return ReadFileChunk.from_filename(filename, start_byte,
-                                           size, callback)
+                                           size, callback,
+                                           enable_callback=False)
 
     def open(self, filename, mode):
         return open(filename, mode)
@@ -583,6 +609,13 @@ class S3Transfer(object):
         if extra_args is None:
             extra_args = {}
         self._validate_all_known_args(extra_args, self.ALLOWED_UPLOAD_ARGS)
+        events = self._client.meta.events
+        events.register_first('request-created.s3',
+                              disable_upload_callbacks,
+                              unique_id='s3upload-callback-disable')
+        events.register_last('request-created.s3',
+                             enable_upload_callbacks,
+                             unique_id='s3upload-callback-enable')
         if self._osutil.get_file_size(filename) >= \\
                 self._config.multipart_threshold:
             self._multipart_upload(filename, bucket, key, callback, extra_args)
diff --git a/tests/integration/test_s3.py b/tests/integration/test_s3.py
index 1edc1167..2fe28db3 100644
--- a/tests/integration/test_s3.py
+++ b/tests/integration/test_s3.py
@@ -22,6 +22,7 @@ import string
 
 from tests import unittest, unique_id
 from botocore.compat import six
+from botocore.client import Config
 
 import boto3.session
 import boto3.s3.transfer
@@ -343,6 +344,27 @@ class TestS3Transfers(unittest.TestCase):
         # of the file we uploaded.
         self.assertEqual(self.amount_seen, 20 * 1024 * 1024)
 
+    def test_callback_called_once_with_sigv4(self):
+        # Verify #98, where the callback was being invoked
+        # twice when using signature version 4.
+        self.amount_seen = 0
+        lock = threading.Lock()
+        def progress_callback(amount):
+            with lock:
+                self.amount_seen += amount
+
+        client = self.session.client(
+            's3', self.region,
+            config=Config(signature_version='s3v4'))
+        transfer = boto3.s3.transfer.S3Transfer(client)
+        filename = self.files.create_file_with_size(
+            '10mb.txt', filesize=10 * 1024 * 1024)
+        transfer.upload_file(filename, self.bucket_name,
+                             '10mb.txt', callback=progress_callback)
+        self.addCleanup(self.delete_object, '10mb.txt')
+
+        self.assertEqual(self.amount_seen, 10 * 1024 * 1024)
+
     def test_can_send_extra_params_on_upload(self):
         transfer = self.create_s3_transfer()
         filename = self.files.create_file_with_size('foo.txt', filesize=1024)
diff --git a/tests/unit/s3/test_transfer.py b/tests/unit/s3/test_transfer.py
index b020c29f..fb778fde 100644
--- a/tests/unit/s3/test_transfer.py
+++ b/tests/unit/s3/test_transfer.py
@@ -30,6 +30,7 @@ from boto3.s3.transfer import MultipartDownloader, MultipartUploader
 from boto3.s3.transfer import ShutdownQueue
 from boto3.s3.transfer import QueueShutdownError
 from boto3.s3.transfer import random_file_extension
+from boto3.s3.transfer import disable_upload_callbacks, enable_upload_callbacks
 
 
 class InMemoryOSLayer(OSUtils):
@@ -99,7 +100,8 @@ class TestOSUtils(unittest.TestCase):
     def test_open_file_chunk_reader(self):
         with mock.patch('boto3.s3.transfer.ReadFileChunk') as m:
             OSUtils().open_file_chunk_reader('myfile', 0, 100, None)
-            m.from_filename.assert_called_with('myfile', 0, 100, None)
+            m.from_filename.assert_called_with('myfile', 0, 100,
+                                               None, enable_callback=False)
 
     def test_open_file(self):
         fileobj = OSUtils().open(os.path.join(self.tempdir, 'foo'), 'w')
@@ -199,6 +201,23 @@ class TestReadFileChunk(unittest.TestCase):
 
         self.assertEqual(amounts_seen, [1, 1, 1])
 
+    def test_callback_can_be_disabled(self):
+        filename = os.path.join(self.tempdir, 'foo')
+        with open(filename, 'wb') as f:
+            f.write(b'abc')
+        callback_calls = []
+
+        def callback(amount):
+            callback_calls.append(amount)
+
+        chunk = ReadFileChunk.from_filename(
+            filename, start_byte=0, chunk_size=3, callback=callback)
+        chunk.disable_callback()
+        # Now reading from the ReadFileChunk should not invoke
+        # the callback.
+        chunk.read()
+        self.assertEqual(callback_calls, [])
+
     def test_file_chunk_supports_context_manager(self):
         filename = os.path.join(self.tempdir, 'foo')
         with open(filename, 'wb') as f:
@@ -472,6 +491,22 @@ class TestS3Transfer(unittest.TestCase):
     def tearDown(self):
         self.random_file_patch.stop()
 
+    def test_callback_handlers_register_on_put_item(self):
+        osutil = InMemoryOSLayer({'smallfile': b'foobar'})
+        transfer = S3Transfer(self.client, osutil=osutil)
+        transfer.upload_file('smallfile', 'bucket', 'key')
+        events = self.client.meta.events
+        events.register_first.assert_called_with(
+            'request-created.s3',
+            disable_upload_callbacks,
+            unique_id='s3upload-callback-disable',
+        )
+        events.register_last.assert_called_with(
+            'request-created.s3',
+            enable_upload_callbacks,
+            unique_id='s3upload-callback-enable',
+        )
+
     def test_upload_below_multipart_threshold_uses_put_object(self):
         fake_files = {
             'smallfile': b'foobar',
@@ -640,3 +675,52 @@ class TestRandomFileExtension(unittest.TestCase):
     def test_has_proper_length(self):
         self.assertEqual(
             len(random_file_extension(num_digits=4)), 4)
+
+
+class TestCallbackHandlers(unittest.TestCase):
+    def setUp(self):
+        self.request = mock.Mock()
+
+    def test_disable_request_on_put_object(self):
+        disable_upload_callbacks(self.request,
+                                 'PutObject')
+        self.request.body.disable_callback.assert_called_with()
+
+    def test_disable_request_on_upload_part(self):
+        disable_upload_callbacks(self.request,
+                                 'UploadPart')
+        self.request.body.disable_callback.assert_called_with()
+
+    def test_enable_object_on_put_object(self):
+        enable_upload_callbacks(self.request,
+                                 'PutObject')
+        self.request.body.enable_callback.assert_called_with()
+
+    def test_enable_object_on_upload_part(self):
+        enable_upload_callbacks(self.request,
+                                 'UploadPart')
+        self.request.body.enable_callback.assert_called_with()
+
+    def test_dont_disable_if_missing_interface(self):
+        del self.request.body.disable_callback
+        disable_upload_callbacks(self.request,
+                                 'PutObject')
+        self.assertEqual(self.request.body.method_calls, [])
+
+    def test_dont_enable_if_missing_interface(self):
+        del self.request.body.enable_callback
+        enable_upload_callbacks(self.request,
+                                'PutObject')
+        self.assertEqual(self.request.body.method_calls, [])
+
+    def test_dont_disable_if_wrong_operation(self):
+        disable_upload_callbacks(self.request,
+                                 'OtherOperation')
+        self.assertFalse(
+            self.request.body.disable_callback.called)
+
+    def test_dont_enable_if_wrong_operation(self):
+        enable_upload_callbacks(self.request,
+                                'OtherOperation')
+        self.assertFalse(
+            self.request.body.enable_callback.called)
['boto3/s3/transfer.py', 'tests/unit/s3/test_transfer.py', 'tests/integration/test_s3.py']