summaryrefslogtreecommitdiffstats
path: root/aconf/transaction/base.lua
blob: cda7ed69b5a62dd8cb2afc3d2096908806c9c51d (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
--[[
Copyright (c) 2012-2015 Kaarle Ritvanen
See LICENSE file for license details
--]]

local M = {}

local err = require('aconf.error')

local object = require('aconf.object')
local class = object.class

local address = require('aconf.path.address')
local util = require('aconf.util')

-- TODO each transaction backend (i.e. persistence manager or
-- transaction proper) should be implemented as a thread or have its
-- internal state stored in shared storage (with appropriate locking)


local generation = 0
local function gen_number()
   generation = generation + 1
   return generation
end


M.TransactionBackend = class()

function M.TransactionBackend:init() self.mod_time = {} end

function M.TransactionBackend:get_if_older(path, timestamp)
   local value, ts = self:get(path)
   if ts > timestamp then err.raise('conflict', path) end
   return value, ts
end

function M.TransactionBackend:set(path, value)
   self:set_multiple{{path, value}}
end

function M.TransactionBackend:set_multiple(mods)
   -- TODO delegate to PM backends?
   local timestamp = gen_number()
   local effective = {}

   for _, mod in ipairs(mods) do
      local path, value = table.unpack(mod)

      if type(value) == 'table' or type(
	 self:get(path)
      ) == 'table' or self:get(path) ~= value then

	 table.insert(effective, mod)
	 self.mod_time[path] = timestamp
      end
   end

   self:_set_multiple(effective)
end

-- TODO should be atomic, mutex with set_multiple
function M.TransactionBackend:comp_and_setm(accessed, mods)
   local errors = err.ErrorDict()
   for path, timestamp in pairs(accessed) do
      errors:collect(self.get_if_older, self, path, timestamp)
   end
   errors:raise()

   self:set_multiple(mods)
end



local function remove_list_value(list, value)
   value = tostring(value)

   for i, v in ipairs(list) do
      if tostring(v) == value then
	 table.remove(list, i)
	 return
      end
   end
end


M.Transaction = class(M.TransactionBackend)

function M.Transaction:init(backend)
   object.super(self, M.Transaction):init()
   self.backend = backend
   self:reset()
end

function M.Transaction:reset()
   self.started = gen_number()
   self.access_time = {}

   self.added = {}
   self.modified = {}
   self.deleted = {}
end

function M.Transaction:get(path)
   if self.deleted[path] then return nil, self.mod_time[path] end
   for _, tbl in ipairs{self.added, self.modified} do
      if tbl[path] ~= nil then
	 return util.copy(tbl[path]), self.mod_time[path]
      end
   end

   local value, timestamp = self.backend:get_if_older(path, self.started)
   self.access_time[path] = timestamp
   return value, timestamp
end

function M.Transaction:expand(path)
   local prefix = {}
   path = address.split(path)

   while path[1] do
      local comp = path[1]
      table.remove(path, 1)

      if comp == address.wildcard then
	 local p = address.join('/', table.unpack(prefix))
	 local res = {}

	 local children = self:get(p) or {}
	 table.sort(children)
	 for _, child in ipairs(children) do
	    util.extend(
	       res, self:expand(address.join(p, child, table.unpack(path)))
	    )
	 end

	 return res
      end

      table.insert(prefix, comp)
   end

   return {address.join('/', table.unpack(prefix))}
end

function M.Transaction:_set_multiple(mods)

   local function set(path, value, new)
      local delete = value == nil

      if self.added[path] == nil and (not new or self.deleted[path]) then
	 self.modified[path] = value
	 self.deleted[path] = delete
      else self.added[path] = value end
   end

   for _, mod in ipairs(mods) do
      local path, value = table.unpack(mod)

      local ppath = address.parent(path)
      local parent = self:get(ppath)
      if parent == nil then
	 parent = {}
	 self:set(ppath, parent)
      end

      local name = address.name(path)
      local old = self:get(path)

      local is_table = type(value) == 'table'
      local delete = value == nil

      if delete then self:check_deleted(path) end

      if type(old) == 'table' then
	 if delete then
	    for _, child in ipairs(old) do
	       self:set(address.join(path, child))
	    end
	 elseif is_table then return
	 elseif #old > 0 then
	    error('Cannot assign a primitive value to non-leaf node '..path)
	 end
      end

      if is_table then value = {} end
      set(path, value, old == nil)

      local function set_parent()
	 set(ppath, parent)
	 self.mod_time[ppath] = self.mod_time[path]
      end

      if old == nil and not delete then
	 table.insert(parent, name)
	 set_parent()
      elseif old ~= nil and delete then
	 remove_list_value(parent, name)
	 set_parent()
      end
   end
end

function M.Transaction:check_deleted(path) end

function M.Transaction:commit()
   local mods = {}
   local handled = {}

   local function insert(path, value)
      assert(not handled[path])
      table.insert(mods, {path, value})
      handled[path] = true
   end

   local function insert_add(path)
      if not handled[path] then
	 local pp = address.parent(path)
	 if self.added[pp] then insert_add(pp) end
	 insert(path, self.added[path])
      end
   end

   local function insert_del(path)
      if not handled[path] then
	 local value = self.backend:get(path)
	 if type(value) == 'table' then
	    for _, child in ipairs(value) do
	       local cp = address.join(path, child)
	       assert(self.deleted[cp])
	       insert_del(cp)
	    end
	 end
	 insert(path)
      end
   end

   for path, deleted in pairs(self.deleted) do
      if deleted then insert_del(path) end
   end

   for path, value in pairs(self.modified) do
      if type(value) ~= 'table' then insert(path, value) end
   end

   for path, _ in pairs(self.added) do insert_add(path) end

   self.backend:comp_and_setm(self.access_time, mods)

   self:reset()
end


return M