lib/mogilefs/mogilefs.rb

   1 require 'socket'
   2 require 'timeout'
   3
   4 require 'mogilefs/client'
   5 require 'mogilefs/nfsfile'
   6 require 'mogilefs/util'
   7
   8 ##
   9 # Timeout error class.
  10
  11 class MogileFS::Timeout < Timeout::Error; end
  12
  13 ##
  14 # MogileFS File manipulation client.
  15
  16 class MogileFS::MogileFS < MogileFS::Client
  17
  18   include MogileFS::Util
  19   include MogileFS::Bigfile
  20
  21   ##
  22   # The path to the local MogileFS mount point if you are using NFS mode.
  23
  24   attr_reader :root
  25
  26   ##
  27   # The domain of keys for this MogileFS client.
  28
  29   attr_reader :domain
  30
  31   ##
  32   # The timeout for get_file_data.  Defaults to five seconds.
  33
  34   attr_accessor :get_file_data_timeout
  35
  36   ##
  37   # Creates a new MogileFS::MogileFS instance.  +args+ must include a key
  38   # :domain specifying the domain of this client.  A key :root will be used to
  39   # specify the root of the NFS file system.
  40
  41   def initialize(args = {})
  42     @domain = args[:domain]
  43     @root = args[:root]
  44
  45     @get_file_data_timeout = 5
  46
  47     raise ArgumentError, "you must specify a domain" unless @domain
  48
  49     if @backend = args[:db_backend]
  50       @readonly = true
  51     else
  52       super
  53     end
  54   end
  55
  56   ##
  57   # Enumerates keys starting with +key+.
  58
  59   def each_key(prefix)
  60     after = nil
  61
  62     keys, after = list_keys prefix
  63
  64     until keys.nil? or keys.empty? do
  65       keys.each { |k| yield k }
  66       keys, after = list_keys prefix, after
  67     end
  68
  69     return nil
  70   end
  71
  72   ##
  73   # Retrieves the contents of +key+.
  74
  75   def get_file_data(key, &block)
  76     paths = get_paths key
  77
  78     return nil unless paths
  79
  80     paths.each do |path|
  81       next unless path
  82       case path
  83       when /^http:\/\// then
  84         begin
  85           sock = http_get_sock(URI.parse(path))
  86           return block_given? ? yield(sock) : sock.read
  87         rescue MogileFS::Timeout, Errno::ECONNREFUSED,
  88                EOFError, SystemCallError
  89           next
  90         end
  91       else
  92         next unless File.exist? path
  93         return File.read(path)
  94       end
  95     end
  96
  97     return nil
  98   end
  99
 100   ##
 101   # Get the paths for +key+.
 102
 103   def get_paths(key, noverify = true, zone = nil)
 104     noverify = noverify ? 1 : 0
 105     res = @backend.get_paths(:domain => @domain, :key => key,
 106                              :noverify => noverify, :zone => zone)
 107     paths = (1..res['paths'].to_i).map { |i| res["path#{i}"] }
 108     return paths if paths.empty?
 109     return paths if paths.first =~ /^http:\/\//
 110     return paths.map { |path| File.join @root, path }
 111   end
 112
 113   ##
 114   # Creates a new file +key+ in +klass+.  +bytes+ is currently unused.
 115   #
 116   # The +block+ operates like File.open.
 117
 118   def new_file(key, klass, bytes = 0, &block) # :yields: file
 119     raise MogileFS::ReadOnlyError if readonly?
 120
 121     res = @backend.create_open(:domain => @domain, :class => klass,
 122                                :key => key, :multi_dest => 1)
 123
 124     dests = nil
 125
 126     if res.include? 'dev_count' then # HACK HUH?
 127       dests = (1..res['dev_count'].to_i).map do |i|
 128         [res["devid_#{i}"], res["path_#{i}"]]
 129       end
 130     else
 131       # 0x0040:  d0e4 4f4b 2064 6576 6964 3d31 2666 6964  ..OK.devid=1&fid
 132       # 0x0050:  3d33 2670 6174 683d 6874 7470 3a2f 2f31  =3&path=http://1
 133       # 0x0060:  3932 2e31 3638 2e31 2e37 323a 3735 3030  92.168.1.72:7500
 134       # 0x0070:  2f64 6576 312f 302f 3030 302f 3030 302f  /dev1/0/000/000/
 135       # 0x0080:  3030 3030 3030 3030 3033 2e66 6964 0d0a  0000000003.fid..
 136
 137       dests = [[res['devid'], res['path']]]
 138     end
 139
 140     dest = dests.first
 141     devid, path = dest
 142
 143     case path
 144     when nil, '' then
 145       raise EmptyPathError
 146     when /^http:\/\// then
 147       MogileFS::HTTPFile.open(self, res['fid'], path, devid, klass, key,
 148                               dests, bytes, &block)
 149     else
 150       MogileFS::NFSFile.open(self, res['fid'], path, devid, klass, key, &block)
 151     end
 152   end
 153
 154   ##
 155   # Copies the contents of +file+ into +key+ in class +klass+.  +file+ can be
 156   # either a file name or an object that responds to #read.
 157
 158   def store_file(key, klass, file)
 159     raise MogileFS::ReadOnlyError if readonly?
 160
 161     new_file key, klass do |mfp|
 162       if file.respond_to? :sysread then
 163         return sysrwloop(file, mfp)
 164       else
 165         if File.size(file) > 0x10000 # Bigass file, handle differently
 166           mfp.bigfile = file
 167           return
 168         else
 169           return File.open(file) { |fp| sysrwloop(fp, mfp) }
 170         end
 171       end
 172     end
 173   end
 174
 175   ##
 176   # Stores +content+ into +key+ in class +klass+.
 177
 178   def store_content(key, klass, content)
 179     raise MogileFS::ReadOnlyError if readonly?
 180
 181     new_file key, klass do |mfp|
 182       mfp << content
 183     end
 184
 185     return content.length
 186   end
 187
 188   ##
 189   # Removes +key+.
 190
 191   def delete(key)
 192     raise MogileFS::ReadOnlyError if readonly?
 193
 194     @backend.delete :domain => @domain, :key => key
 195   end
 196
 197   ##
 198   # Sleeps +duration+.
 199
 200   def sleep(duration)
 201     @backend.sleep :duration => duration
 202   end
 203
 204   ##
 205   # Renames a key +from+ to key +to+.
 206
 207   def rename(from, to)
 208     raise MogileFS::ReadOnlyError if readonly?
 209
 210     @backend.rename :domain => @domain, :from_key => from, :to_key => to
 211     nil
 212   end
 213
 214   ##
 215   # Returns the size of +key+.
 216   def size(key)
 217     @backend.respond_to?(:_size) and return @backend._size(domain, key)
 218     paths = get_paths(key) or return nil
 219     paths_size(paths)
 220   end
 221
 222   def paths_size(paths)
 223     paths.each do |path|
 224       next unless path
 225       case path
 226       when /^http:\/\// then
 227         begin
 228           url = URI.parse path
 229
 230           res = timeout @get_file_data_timeout, MogileFS::Timeout do
 231             s = TCPSocket.new(url.host, url.port)
 232             s.syswrite("HEAD #{url.request_uri} HTTP/1.0\r\n\r\n")
 233             s.sysread(4096)
 234           end
 235           if cl = /^Content-Length:\s*(\d+)/i.match(res)
 236             return cl[1].to_i
 237           end
 238           next
 239         rescue MogileFS::Timeout, Errno::ECONNREFUSED,
 240                EOFError, SystemCallError
 241           next
 242         end
 243       else
 244         next unless File.exist? path
 245         return File.size(path)
 246       end
 247     end
 248
 249     nil
 250   end
 251
 252   ##
 253   # Lists keys starting with +prefix+ follwing +after+ up to +limit+.  If
 254   # +after+ is nil the list starts at the beginning.
 255
 256   def list_keys(prefix, after = nil, limit = 1000, &block)
 257     if @backend.respond_to?(:_list_keys)
 258       return @backend._list_keys(domain, prefix, after, limit, &block)
 259     end
 260
 261     res = begin
 262       @backend.list_keys(:domain => domain, :prefix => prefix,
 263                          :after => after, :limit => limit)
 264     rescue MogileFS::Backend::NoneMatchError
 265       return nil
 266     end
 267
 268     keys = (1..res['key_count'].to_i).map { |i| res["key_#{i}"] }
 269     if block_given?
 270       # emulate the MogileFS::Mysql interface, slowly...
 271       keys.each do |key|
 272         paths = get_paths(key) or next
 273         length = paths_size(paths) or next
 274         yield key, length, paths.size
 275       end
 276     end
 277
 278     return keys, res['next_after']
 279   end
 280
 281   protected
 282
 283     def http_get_sock(uri)
 284       sock = nil
 285       timeout @get_file_data_timeout, MogileFS::Timeout do
 286         sock = TCPSocket.new(uri.host, uri.port)
 287         sock.sync = true
 288         sock.syswrite("GET #{uri.request_uri} HTTP/1.0\r\n\r\n")
 289         buf = sock.recv(4096, Socket::MSG_PEEK)
 290         head, body = buf.split(/\r\n\r\n/, 2)
 291         head = sock.recv(head.size + 4)
 292       end
 293
 294       sock
 295     end # def http_get_sock
 296
 297 end
 298