Write logic refactor.

- No longer have to read whole file into memory in order to update.
- We're not done, yet. This still needs to be tested, and we're still investigating what could be causign the 512-byte error that people are seeing.

Author: dsoprea

gdrivefs/gdfs/gdfuse.py

@@ -610,8 +610,6 @@ def truncate(self, filepath, length, fh=None):
                                      "with ID (%d) (truncate)." % (fh))
                 raise FuseOSError(EIO)
 
-            opened_file.reset_state()
-
             entry_id = opened_file.entry_id
             cache = EntryCache.get_instance().cache
 
@@ -622,6 +620,8 @@ def truncate(self, filepath, length, fh=None):
                                      "ID [%s] for truncate with FH." % 
                                      (entry_id))
                 raise
+
+            opened_file.truncate(length)
         else:
             (entry, path, filename) = get_entry_or_raise(filepath)
 
@@ -631,6 +631,9 @@ def truncate(self, filepath, length, fh=None):
             _logger.exception("Could not truncate entry [%s]." % (entry))
             raise FuseOSError(EIO)
 
+# TODO(dustin): It would be a lot quicker if we truncate our temporary file 
+#               here, and make sure its mtime matches.
+
         # We don't need to update our internal representation of the file (just 
         # our file-handle and its related buffering).
 

gdrivefs/gdfs/opened_file.py

@@ -1,6 +1,8 @@
 import logging
 import resource
 import re
+import time
+import os
 
 from errno import *
 from threading import Lock, RLock
@@ -17,7 +19,6 @@
 from gdrivefs.cache.volume import PathRelations, EntryCache, path_resolver, \
                                   CLAUSE_ID, CLAUSE_ENTRY
 from gdrivefs.gdtool.drive import drive_proxy
-from gdrivefs.general.buffer_segments import BufferSegments
 
 _static_log = logging.getLogger().getChild('(OF)')
 
@@ -34,19 +35,6 @@ def get_temp_filepath(normalized_entry, mime_type):
     return ("%s/local/%s" % (temp_path, temp_filename))
 
 
-
-# TODO(dustin): LCM runs in a greenlet pool. When we open a file that needs the
-#               existing data for a file (read, append), a switch is done to an
-#               LCM worker. If the data is absent or faulted, download the
-#               content. Then, switch back.
-
-class LocalCopyManager(object):
-    """Manages local copies of files."""
-    
-#    def 
-    pass
-
-
 class OpenedManager(object):
     """Manages all of the currently-open files."""
 
@@ -296,8 +284,16 @@ def __init__(self, entry_id, path, filename, is_hidden, mime_type):
 
         self.reset_state()
 
+        try:
+            entry = self.__get_entry_or_raise()
+        except:
+            self.__log.exception("Could not get entry with ID [%s] for "
+                                 "write-flush." % (self.__entry_id))
+            raise
+
+        self.__temp_filepath = get_temp_filepath(entry, self.mime_type)
+
     def reset_state(self):
-        self.__buffer = None
         self.__is_loaded = False
         self.__is_dirty = False
 
@@ -348,7 +344,7 @@ def __load_base_from_remote(self):
 
         self.__log.debug("Ensuring local availability of [%s]." % (entry))
 
-        temp_file_path = get_temp_filepath(entry, self.mime_type)
+        temp_filepath = get_temp_filepath(entry, self.mime_type)
 
         self.__log.debug("__load_base_from_remote about to download.")
 
@@ -358,7 +354,7 @@ def __load_base_from_remote(self):
 
             self.__log.info("Attempting local cache update of file [%s] for "
                             "entry [%s] and mime-type [%s]." % 
-                            (temp_file_path, entry, self.mime_type))
+                            (temp_filepath, entry, self.mime_type))
 
             if entry.requires_mimetype:
                 length = DisplacedFile.file_size
@@ -367,11 +363,11 @@ def __load_base_from_remote(self):
                     d = DisplacedFile(entry)
                     stub_data = d.deposit_file(self.mime_type)
 
-                    with file(temp_file_path, 'w') as f:
+                    with file(temp_filepath, 'w') as f:
                         f.write(stub_data)
                 except:
                     self.__log.exception("Could not deposit to file [%s] from "
-                                         "entry [%s]." % (temp_file_path, 
+                                         "entry [%s]." % (temp_filepath, 
                                                           entry))
                     raise
 
@@ -382,9 +378,10 @@ def __load_base_from_remote(self):
                 self.__log.info("Executing the download.")
 
                 try:
-# TODO(dustin): We're not inheriting an existing file (same mtime, same size).
+# TODO(dustin): Confirm that this will inherit an existing file (same mtime, 
+#               same size).
                     result = drive_proxy('download_to_local', 
-                                         output_file_path=temp_file_path,
+                                         output_file_path=temp_filepath,
                                          normalized_entry=entry,
                                          mime_type=self.mime_type)
 
@@ -401,45 +398,6 @@ def __load_base_from_remote(self):
                             "__is_loaded= [%s]" % 
                             (cache_fault, self.__is_loaded))
 
-            # We've either not loaded it, yet, or it has changed.
-            if cache_fault or not self.__is_loaded:
-                with self.__class__.__update_lock:
-                    self.__log.info("Checking queued items for fault.")
-
-                    if cache_fault:
-                        if self.__is_dirty:
-                            self.__log.error("Entry [%s] has been changed. "
-                                             "Forcing buffer updates, and "
-                                             "clearing uncommitted updates." % 
-                                             (entry))
-                        else:
-                            self.__log.debug("Entry [%s] has changed. "
-                                             "Updating buffers." % (entry))
-
-                    self.__log.debug("Loading buffers.")
-
-                    with open(temp_file_path, 'rb') as f:
-                        # Read the locally cached file in.
-
-                        try:
-# TODO(dustin): Our accounting is broken when it comes to loading and/or update-tracking. If we have a guarantee thawrites only appear in sequence and in increasing order, we can dump BufferSegments.
-
-# TODO(dustin): This is the source of:
-# 1) An enormous slowdown where we first have to write the data, and then have to read it back.
-# 2) An enormous resource burden.
-                            data = f.read()
-
-                            read_blocksize = Conf.get('default_buffer_read_blocksize')
-                            self.__buffer = BufferSegments(data, read_blocksize)
-                        except:
-                            self.__log.exception("Could not read current cached "
-                                                 "file into buffer.")
-                            raise
-
-                        self.__is_dirty = False
-
-                    self.__is_loaded = True
-
         self.__log.debug("__load_base_from_remote complete.")
         return cache_fault
 
@@ -454,15 +412,45 @@ def add_update(self, offset, data):
             self.__load_base_from_remote()
         except:
             self.__log.exception("Could not load entry to local cache [%s]." % 
-                                 (self.temp_file_path))
+                                 (self.__temp_filepath))
             raise
 
         self.__log.debug("Base loaded for add_update.")
 
         with self.__class__.__update_lock:
-            self.__buffer.apply_update(offset, data)
+            with open(self.__temp_filepath, 'r+') as f:
+                f.seek(offset)
+                f.write(data)
+
             self.__is_dirty = True
 
+    @dec_hint(['length'], ['length'], 'OF')
+    def truncate(self, length):
+        try:
+            self.__load_base_from_remote()
+        except:
+            self.__log.exception("Could not load entry to local cache [%s]." % 
+                                 (self.__temp_filepath))
+            raise
+
+        self.__log.debug("Base loaded for truncate.")
+
+        entry = self.__get_entry_or_raise()
+
+        with self.__class__.__update_lock:
+            with open(self.__temp_filepath, 'r+') as f:
+                f.truncate(length)
+
+            gd_mtime_epoch = time.mktime(
+                                entry.modified_date.timetuple())
+
+# TODO(dustin): Shouldn't we be taking the first time component (atime?) from the entry?
+            os.utime(self.__temp_filepath, (time.time(), gd_mtime_epoch))
+
+            self.__is_dirty = True
+
+# TODO(dustin): We still have to make sure the mtime is set to match
+
     @dec_hint(prefix='OF')
     def flush(self):
         """The OS wants to effect any changes made to the file."""
@@ -472,81 +460,54 @@ def flush(self):
         entry = self.__get_entry_or_raise()
         cache_fault = self.__load_base_from_remote()
 
+# TODO(dustin): We need to be able to do updates for separate files in 
+#               parallel.
         with self.__class__.__update_lock:
             if self.__is_dirty is False:
                 self.__log.debug("Flush will be skipped because there are no "
                                  "changes.")
-# TODO: Raise an exception?
                 return
 
-            # Write back out to the temporary file.
-
-            self.__log.debug("Writing buffer to temporary file.")
-# TODO: Make sure to uncache the temp data if self.temp_file_path is not None.
-
-            mime_type = self.mime_type
-
-            # If we've already opened a work file, use it. Else, use a 
-            # temporary file that we'll close at the end of the method.
-            if self.__is_loaded:
-                is_temp = False
-
-                temp_file_path = get_temp_filepath(entry, mime_type)
-                                                   
-                with file(temp_file_path, 'w') as f:
-                    for block in self.__buffer.read():
-                        f.write(block)
-                                                   
-                write_filepath = temp_file_path
-            else:
-                is_temp = True
-            
-                with NamedTemporaryFile(delete=False) as f:
-                    write_filepath = f.name
-                    for block in self.__buffer.read():
-                        f.write(block)
-
             # Push to GD.
 
+#            os.stat(self.__temp
+
             self.__log.debug("Pushing (%d) bytes for entry with ID from [%s] "
                              "to GD for file-path [%s]." % 
-                             (self.__buffer.length, entry.id, write_filepath))
-
-#            print("Sending updates.")
+                             (self.__buffer.length, entry.id, self.__temp_filepath))
 
-# TODO: Update mtime?
+# TODO: Will this automatically update mtime?
+# TODO(dustin): We need to be able to update individual slices of the file 
+#               (maybe only if we've affected less than X% of the file).
             try:
                 entry = drive_proxy('update_entry', 
                                     normalized_entry=entry, 
                                     filename=entry.title, 
-                                    data_filepath=write_filepath, 
-                                    mime_type=mime_type, 
+                                    data_filepath=self.__temp_filepath, 
+                                    mime_type=self.mime_type, 
                                     parents=entry.parents, 
                                     is_hidden=self.__is_hidden)
             except:
                 self.__log.exception("Could not localize displaced file with "
                                      "entry having ID [%s]." % (entry.id))
                 raise
 
-            if not is_temp:
-                unlink(write_filepath)
-            else:
-                # Update the write-cache file to the official mtime. We won't 
-                # redownload it on the next flush if it wasn't changed, 
-                # elsewhere.
+            # Update the write-cache file to the official mtime. We won't 
+            # redownload it on the next flush if it wasn't changed, 
+            # elsewhere.
 
-                self.__log.debug("Updating local write-cache file to official "
-                                 "mtime [%s]." % (entry.modified_date_epoch))
+            self.__log.debug("Updating local write-cache file to official "
+                             "mtime [%s]." % (entry.modified_date_epoch))
 
-                try:
-                    utime(write_filepath, (entry.modified_date_epoch, 
-                                            entry.modified_date_epoch))
-                except:
-                    self.__log.exception("Could not update mtime of write-"
-                                         "cache [%s] for entry with ID [%s], "
-                                         "post-flush." % 
-                                         (entry.modified_date_epoch, entry.id))
-                    raise
+            try:
+                utime(self.__temp_filepath, (entry.modified_date_epoch, 
+                                             entry.modified_date_epoch))
+            except:
+                self.__log.exception("Could not update mtime of write-"
+                                     "cache [%s] for entry with ID [%s], "
+                                     "post-flush." % 
+                                     (entry.modified_date_epoch, entry.id))
+                raise
 
         # Immediately update our current cached entry.
 

gdrivefs/gdtool/drive.py

@@ -367,7 +367,8 @@ def download_to_local(self, output_file_path, normalized_entry, mime_type,
             # Use the cache. It's fine.
 
             self.__log.debug("File retrieved from the previously downloaded, "
-                            "still-current file.")
+                             "still-current file.")
+
             return (stat_info.st_size, False)
 
         # Go and get the file.
@@ -377,42 +378,6 @@ def download_to_local(self, output_file_path, normalized_entry, mime_type,
 
         url = normalized_entry.download_links[mime_type]
 
-#        self.__log.debug("Downloading file from [%s]." % (url))
-#
-#        try:
-#            data_tuple = authed_http.request(url)
-#        except:
-#            self.__log.exception("Could not download entry with ID [%s], type "
-#                              "[%s], and URL [%s]." % (normalized_entry.id, 
-#                                                       mime_type, url))
-#            raise
-#
-#        (response_headers, data) = data_tuple
-#
-#        # Throw a log-item if we see any "Range" response-headers. If GD ever
-#        # starts supporting "Range" headers, we'll be able to write smarter 
-#        # download mechanics (resume, etc..).
-#
-#        r = re.compile('Range')
-#        range_found = [("%s: %s" % (k, v)) for k, v 
-#                                           in response_headers.iteritems() 
-#                                           if r.match(k)]
-#        if range_found:
-#            self.__log.info("GD has returned Range-related headers: %s" % 
-#                            (", ".join(found)))
-#
-#        self.__log.info("Downloaded file is (%d) bytes. Writing to [%s]." % 
-#                        (len(data), output_file_path))
-#
-#        try:
-#            with open(output_file_path, 'wb') as f:
-#                f.write(data)
-#        except:
-#            self.__log.exception("Could not cached downloaded file. Skipped.")
-#
-#        else:
-#            self.__log.info("File written to cache successfully.")
-
         with open(output_file_path, 'wb') as f:
             downloader = gdrivefs.gdtool.chunked_download.ChunkedDownload(
                             f, 
@@ -424,12 +389,7 @@ def download_to_local(self, output_file_path, normalized_entry, mime_type,
                 if done is True:
                     break
 
-        try:
-            utime(output_file_path, (time.time(), gd_mtime_epoch))
-        except:
-            self.__log.exception("Could not set time on [%s]." % 
-                                 (output_file_path))
-            raise
+        utime(output_file_path, (time.time(), gd_mtime_epoch))
 
         return (total_size, True)