Class: LS4::DataServerService

Inherits:
Service show all
Defined in:
lib/ls4/service/data_server.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from Service

init

Methods included from EventBus::SingletonMixin

#ebus_bind!, #ebus_connect, extended

Methods included from EventBus::BusMixin

#ebus_all_slots, #ebus_disconnect!

Methods included from EventBus::DeclarerBase::Methods

#connect, #ebus_all_slots, #ebus_call_log, #ebus_call_slots, #ebus_signal_error, #ebus_signal_log, #ebus_signal_slots

Methods included from EventBus::DeclarerBase

#call_slot, #signal_slot

Constructor Details

#initialize ⇒ DataServerService

Returns a new instance of DataServerService.



22
23
24
25
26
27
28
 
# File 'lib/ls4/service/data_server.rb', line 22

def initialize
	@self_nid = ConfigBus.self_nid
	@self_rsids = ConfigBus.self_rsids
	@stat_cmd_read = 0
	@stat_cmd_write = 0
	@stat_cmd_delete = 0
end

Instance Attribute Details

#stat_cmd_delete ⇒ Object (readonly)

Returns the value of attribute stat_cmd_delete.



147
148
149
 
# File 'lib/ls4/service/data_server.rb', line 147

def stat_cmd_delete
  @stat_cmd_delete
end

#stat_cmd_read ⇒ Object (readonly)

Returns the value of attribute stat_cmd_read.



145
146
147
 
# File 'lib/ls4/service/data_server.rb', line 145

def stat_cmd_read
  @stat_cmd_read
end

#stat_cmd_write ⇒ Object (readonly)

Returns the value of attribute stat_cmd_write.



146
147
148
 
# File 'lib/ls4/service/data_server.rb', line 146

def stat_cmd_write
  @stat_cmd_write
end

Instance Method Details

#on_timer ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/ls4/service/data_server.rb', line 127

def on_timer
	nids = []
	@self_rsids.each {|rsid|
		begin
			nids.concat MasterSelectBus.select_master_static(rsid)
		rescue
		end
	}
	done = [@self_nid]
	nids.each {|nid|
		if !done.include?(nid) && !MembershipBus.is_fault(nid)
			session = MembershipBus.get_session_nid(nid)
			SlaveBus.try_replicate(nid, session)
			done << nid
		end
	}
end

#rpc_delete_direct(okey) ⇒ Object 



72
73
74
75
76
77
78
79
80
 
# File 'lib/ls4/service/data_server.rb', line 72

def rpc_delete_direct(okey)
	@stat_cmd_delete += 1
	d = UpdateLogData.new(okey.vtime, okey.key)
	deleted = nil
	UpdateLogBus.append(d.dump) do
		deleted = StorageBus.delete(okey.vtime, okey.key)
	end
	deleted
end

#rpc_exist_direct(okey) ⇒ Object

def rpc_resize_direct(okey, size) # TODO: stat_cmd_resize? # FIXME size field? d = UpdateLogData.new(okey.vtime, okey.key, nil, size) UpdateLogBus.append(d.dump) do StorageBus.resize(okey.vtime, okey.key, size) end nil end



68
69
70
 
# File 'lib/ls4/service/data_server.rb', line 68

def rpc_exist_direct(okey)
	StorageBus.exist(okey.vtime, okey.key)
end

#rpc_get_direct(okey) ⇒ Object 



30
31
32
33
 
# File 'lib/ls4/service/data_server.rb', line 30

def rpc_get_direct(okey)
	@stat_cmd_read += 1
	StorageBus.get(okey.vtime, okey.key)
end

#rpc_read_direct(okey, offset, size) ⇒ Object 



35
36
37
38
 
# File 'lib/ls4/service/data_server.rb', line 35

def rpc_read_direct(okey, offset, size)
	@stat_cmd_read += 1
	StorageBus.read(okey.vtime, okey.key, offset, size)
end

#rpc_replicate_notify(nid) ⇒ Object 



117
118
119
120
121
 
# File 'lib/ls4/service/data_server.rb', line 117

def rpc_replicate_notify(nid)
	session = MembershipBus.get_session_nid(nid)
	SlaveBus.try_replicate(nid, session)
	nil
end

#rpc_replicate_pull(pos, limit) ⇒ Object 



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
 
# File 'lib/ls4/service/data_server.rb', line 82

def rpc_replicate_pull(pos, limit)
	mkeys = []
	msgs = []
	msize = 0
	while true
		raw, npos = UpdateLogBus.get(pos)
		unless raw
			break
		end
		d = UpdateLogData.load(raw)
		# set or delete
		if mkeys.include?(d.key)
			pos = npos
		else
			if d.offset && d.size
				data = StorageBus.read(d.vtime, d.key, d.offset, d.size)
			else
				data = StorageBus.get(d.vtime, d.key)
				mkeys << d.key
			end
			# data may be null => deleted
			if data
				msgs << [d.vtime, d.key, d.offset, data]
				msize += data.size
			else
				# data is deleted
				msgs << [d.vtime, d.key, 0, nil]
			end
			pos = npos
			break if msize > limit
		end
	end
	[pos, msgs]
end

#rpc_set_direct(okey, data) ⇒ Object 



40
41
42
43
44
45
46
47
 
# File 'lib/ls4/service/data_server.rb', line 40

def rpc_set_direct(okey, data)
	@stat_cmd_write += 1
	d = UpdateLogData.new(okey.vtime, okey.key)
	UpdateLogBus.append(d.dump) do
		StorageBus.set(okey.vtime, okey.key, data)
	end
	nil
end

#stat_db_items ⇒ Object 



123
124
125
 
# File 'lib/ls4/service/data_server.rb', line 123

def stat_db_items
	StorageBus.get_items
end