diff --git a/s3fs/core.py b/s3fs/core.py
index 161fef78..a3663c83 100644
--- a/s3fs/core.py
+++ b/s3fs/core.py
@@ -898,13 +898,16 @@ async def _pipe_file(self, path, data, chunksize=50 * 2 ** 20, **kwargs):
 
     async def _put_file(self, lpath, rpath, chunksize=50 * 2 ** 20, **kwargs):
         bucket, key, _ = self.split_path(rpath)
-        if os.path.isdir(lpath) and key:
-            # don't make remote "directory"
-            return
+        if os.path.isdir(lpath):
+            if key:
+                # don't make remote "directory"
+                return
+            else:
+                await self._mkdir(lpath)
         size = os.path.getsize(lpath)
         with open(lpath, "rb") as f0:
             if size < min(5 * 2 ** 30, 2 * chunksize):
-                return await self._call_s3(
+                await self._call_s3(
                     "put_object", Bucket=bucket, Key=key, Body=f0, **kwargs
                 )
             else:
@@ -939,7 +942,9 @@ async def _put_file(self, lpath, rpath, chunksize=50 * 2 ** 20, **kwargs):
                     UploadId=mpu["UploadId"],
                     MultipartUpload={"Parts": parts},
                 )
-        self.invalidate_cache(rpath)
+        while rpath:
+            self.invalidate_cache(rpath)
+            rpath = self._parent(rpath)
 
     async def _get_file(self, rpath, lpath, version_id=None):
         bucket, key, vers = self.split_path(rpath)
diff --git a/s3fs/tests/test_s3fs.py b/s3fs/tests/test_s3fs.py
index c884152b..a6d84568 100644
--- a/s3fs/tests/test_s3fs.py
+++ b/s3fs/tests/test_s3fs.py
@@ -2148,3 +2148,14 @@ def test_list_after_find(s3):
     s3.find("s3://test/2014-01-01.csv")
     after = s3.ls("s3://test")
     assert before == after
+
+
+def test_upload_recursive_to_bucket(s3, tmpdir):
+    # GH#491
+    folders = [os.path.join(tmpdir, d) for d in ["outer", "outer/inner"]]
+    files = [os.path.join(tmpdir, f) for f in ["outer/afile", "outer/inner/bfile"]]
+    for d in folders:
+        os.mkdir(d)
+    for f in files:
+        open(f, "w").write("hello")
+    s3.put(folders[0], "newbucket", recursive=True)