Class: BBFS::ContentServer::FileCopyServer

Inherits:
Object
  • Object
show all
Defined in:
lib/content_server/queue_copy.rb

Overview

Simple copier, gets inputs events (files to copy), requests ack from backup to copy then copies one file.

Instance Method Summary collapse

Constructor Details

#initialize(copy_input_queue, port) ⇒ FileCopyServer

Returns a new instance of FileCopyServer.



24
25
26
27
28
29
30
31
32
33
 
# File 'lib/content_server/queue_copy.rb', line 24

def initialize(copy_input_queue, port)
  # Local simple tcp connection.
  @backup_tcp = Networking::TCPServer.new(port, method(:receive_message))
  @copy_input_queue = copy_input_queue
  # Stores for each checksum, the file source path.
  # TODO(kolman): If there are items in copy_prepare which timeout (don't get ack),
  # resend the ack request.
  @copy_prepare = {}
  @file_streamer = FileStreamer.new(method(:send_chunk))
end

Instance Method Details

#receive_message(addr_info, message) ⇒ Object 



39
40
41
42
43
 
# File 'lib/content_server/queue_copy.rb', line 39

def receive_message(addr_info, message)
  # Add ack message to copy queue.
  Log.info("message received: #{message}")
  @copy_input_queue.push(message)
end

#run ⇒ Object 



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/content_server/queue_copy.rb', line 45

def run()
  threads = []
  threads << @backup_tcp.tcp_thread if @backup_tcp != nil
  threads << Thread.new do
    while true do
      Log.info 'Waiting on copy files events.'
      message_type, message_content = @copy_input_queue.pop

      if message_type == :COPY_MESSAGE
        Log.info "Copy file event: #{message_content}"
        # Prepare source,dest map for copy.
        message_content.instances.each { |key, instance|
          # If not already sending.
          if !@copy_prepare.key?(instance.checksum) || !@copy_prepare[instance.checksum][1]
            @copy_prepare[instance.checksum] = [instance.full_path, false]
            Log.info("Sending ack for: #{instance.checksum}")
            @backup_tcp.send_obj([:ACK_MESSAGE, [instance.checksum, Time.now.to_i]])
          end
        }
      elsif message_type == :ACK_MESSAGE
        # Received ack from backup, copy file if all is good.
        # The timestamp is of local content server! not backup server!
        timestamp, ack, checksum = message_content

        Log.info("Ack (#{ack}) received for: #{checksum}, timestamp: #{timestamp} " \
                 "now: #{Time.now.to_i}")

        # Copy file if ack (does not exists on backup and not too much time passed)
        if ack && (Time.now.to_i - timestamp < Params['ack_timeout'])
          if !@copy_prepare.key?(checksum) || @copy_prepare[checksum][1]
            Log.warning("File was aborted, copied, or started copy just now: #{checksum}")
          else
            path = @copy_prepare[checksum][0]
            Log.info "Streaming file: #{checksum} #{path}."
            @file_streamer.start_streaming(checksum, path)
            # Ack received, setting prepare to true
            @copy_prepare[checksum][1] = true
          end
        else
          Log.debug1("Ack timed out span: #{Time.now.to_i - timestamp} > " \
                     "timeout: #{Params['ack_timeout']}")
        end
      elsif message_type == :COPY_CHUNK_FROM_REMOTE
        checksum = message_content
        @file_streamer.copy_another_chuck(checksum)
      elsif message_type == :COPY_CHUNK
        # We open the message here for printing info and deleting copy_prepare!
        file_checksum, offset, file_size, content, content_checksum = message_content
        Log.info("Send chunk for file #{file_checksum}, offset: #{offset} " \
                 "filesize: #{file_size}.")
        # Blocking send.
        @backup_tcp.send_obj([:COPY_CHUNK, message_content])
        if content.nil? and content_checksum.nil?
          @copy_prepare.delete(file_checksum)
        end
      elsif message_type == :ABORT_COPY
        Log.info("Aborting file copy: #{message_content}")
        if @copy_prepare.key?(message_content)
          Log.info("Aborting: #{@copy_prepare[message_content][0]}")
          @copy_prepare.delete(message_content)
        end
        @file_streamer.abort_streaming(message_content)
      elsif message_type == :RESET_RESUME_COPY
        file_checksum, new_offset = message_content
        Log.info("Resetting/Resuming file (#{file_checksum}) copy to #{new_offset}")
        @file_streamer.reset_streaming(file_checksum, new_offset)
      else
        Log.error("Copy event not supported: #{message_type}")
      end # handle messages here
    end
    Log.error("Should not reach here, loop should continue.")
  end
end

#send_chunk(*arg) ⇒ Object 



35
36
37
 
# File 'lib/content_server/queue_copy.rb', line 35

def send_chunk(*arg)
  @copy_input_queue.push([:COPY_CHUNK, arg])
end