<commit_before>#!/usr/bin/lua

--! Minimalistic CRDT-like shared state structure suitable for mesh networks
--!
--! Copyright (C) 2019-2020  <NAME> <<EMAIL>>
--!
--! This program is free software: you can redistribute it and/or modify
--! it under the terms of the GNU Affero General Public License version 3 as
--! published by the Free Software Foundation.
--!
--! This program is distributed in the hope that it will be useful,
--! but WITHOUT ANY WARRANTY; without even the implied warranty of
--! MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
--! GNU Affero General Public License for more details.
--!
--! You should have received a copy of the GNU Affero General Public License
--! along with this program.  If not, see <http://www.gnu.org/licenses/>.

local fs = require("nixio.fs")
local JSON = require("luci.jsonc")
local nixio = require("nixio")
local uci = require("uci")
local utils = require("lime.utils")

local shared_state = {}
shared_state.DATA_DIR = '/var/shared-state/data/'
shared_state.PERSISTENT_DATA_DIR = '/etc/shared-state/persistent-data/'
shared_state.ERROR_LOCK_FAILED = 165
shared_state.CANDIDATE_NEIGHBORS_BIN = '/usr/bin/shared-state-get_candidates_neigh'

local SharedStateBase = {}

function SharedStateBase:load()
	for key, value in pairs(JSON.parse(self.storageFD:readall()) or {}) do
		self.storage[key] = value
	end
end

function SharedStateBase:lock(maxwait)
	if self.locked then return end
	maxwait = maxwait or 10
	fs.mkdirr(fs.dirname(self.dataFile))
	self.storageFD = nixio.open(
		self.dataFile, nixio.open_flags("rdwr", "creat") )

	for i=1,maxwait do
		if not self.storageFD:lock("tlock") then
			nixio.nanosleep(1)
		else
			self.locked = true
			break
		end
	end

	if not self.locked then
		self.log( "err", self.dataFile, "Failed acquiring lock on data!" )
		os.exit(shared_state.ERROR_LOCK_FAILED)
	end
end

function SharedStateBase:merge(stateSlice)
	self:lock()
	self:load()
	self:_merge(stateSlice)
	self:save()
	self:unlock()
	self:notifyHooks()
end

function SharedStateBase:notifyHooks()
	if self.changed then
		local jsonString = self:toJsonString()
		if not fs.dir(self.hooksDir) then return end
		for hook in fs.dir(self.hooksDir) do
			local cStdin = io.popen(self.hooksDir.."/"..hook, "w")
			cStdin:write(jsonString)
			cStdin:close()
		end
	end
end

function SharedStateBase:save()
	if self.changed then
		local outFd = io.open(self.dataFile, "w")
		outFd:write(self:toJsonString())
		outFd:close()
		outFd = nil
	end
end

function SharedStateBase:httpRequest(url, body)
	local tmpfname = os.tmpname()

	local tmpfd = io.open(tmpfname, "w")
	tmpfd:write(body)
	tmpfd:close()
	tmpfd = nil

	local cmd = "uclient-fetch --no-check-certificate -q -O- --timeout=3 "
	cmd = cmd.."--post-file='"..tmpfname.."' '"..url.."' ; "
	cmd = cmd.."rm -f '"..tmpfname.."'"
	local fd = io.popen(cmd)

	local value = fd:read("*a")
	fd:close()

	return value
end

function SharedStateBase:_sync(urls)
	urls = urls or {}

	if #urls < 1 then
		local uci_cursor = uci:cursor()
		local fixed_candidates =
				uci_cursor:get("shared-state", "options","candidates") or {}
		for _, line in pairs(fixed_candidates) do
			table.insert(
				urls,
				line.."/"..self.dataType )
		end

		io.input(io.popen(shared_state.CANDIDATE_NEIGHBORS_BIN))
		for line in io.lines() do
			table.insert(
				urls,
				self:getSyncUrl(line, self.dataType))
		end
	end

	for _,url in ipairs(urls) do
		local body = self:toJsonString()

		local response = self:httpRequest(url, body)

		if type(response) == "string" and response:len() > 1  then
			local parsedJson = JSON.parse(response)
			if parsedJson then self:_merge(parsedJson) end
		else
			self.log( "debug", "error requesting "..url )
		end
	end
end

function SharedStateBase:sync(urls)
	self:lock()
	self:load()
	self:unlock()
	self:_sync(urls)
	self:lock()
	self:load() -- Take in account changes happened during sync
	self:save()
	self:unlock()
	self:notifyHooks()
end

function SharedStateBase:toJsonString()
	return JSON.stringify(self.storage)
end<commit_msg>shared-state: fix sync not updating keys<commit_after>#!/usr/bin/lua

--! Minimalistic CRDT-like shared state structure suitable for mesh networks
--!
--! Copyright (C) 2019-2020  <NAME> <<EMAIL>>
--!
--! This program is free software: you can redistribute it and/or modify
--! it under the terms of the GNU Affero General Public License version 3 as
--! published by the Free Software Foundation.
--!
--! This program is distributed in the hope that it will be useful,
--! but WITHOUT ANY WARRANTY; without even the implied warranty of
--! MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
--! GNU Affero General Public License for more details.
--!
--! You should have received a copy of the GNU Affero General Public License
--! along with this program.  If not, see <http://www.gnu.org/licenses/>.

local fs = require("nixio.fs")
local JSON = require("luci.jsonc")
local nixio = require("nixio")
local uci = require("uci")
local utils = require("lime.utils")

local shared_state = {}
shared_state.DATA_DIR = '/var/shared-state/data/'
shared_state.PERSISTENT_DATA_DIR = '/etc/shared-state/persistent-data/'
shared_state.ERROR_LOCK_FAILED = 165
shared_state.CANDIDATE_NEIGHBORS_BIN = '/usr/bin/shared-state-get_candidates_neigh'

local SharedStateBase = {}

function SharedStateBase:load(mergeWithCurrentState)
	local onDiskData = JSON.parse(self.storageFD:readall()) or {}
	if mergeWithCurrentState then
		self:_merge(onDiskData)
	else
		for key, value in pairs(onDiskData) do
			self.storage[key] = value
		end
	end
end

function SharedStateBase:lock(maxwait)
	if self.locked then return end
	maxwait = maxwait or 10
	fs.mkdirr(fs.dirname(self.dataFile))
	self.storageFD = nixio.open(
		self.dataFile, nixio.open_flags("rdwr", "creat") )

	for i=1,maxwait do
		if not self.storageFD:lock("tlock") then
			nixio.nanosleep(1)
		else
			self.locked = true
			break
		end
	end

	if not self.locked then
		self.log( "err", self.dataFile, "Failed acquiring lock on data!" )
		os.exit(shared_state.ERROR_LOCK_FAILED)
	end
end

function SharedStateBase:merge(stateSlice)
	self:lock()
	self:load()
	self:_merge(stateSlice)
	self:save()
	self:unlock()
	self:notifyHooks()
end

function SharedStateBase:notifyHooks()
	if self.changed then
		local jsonString = self:toJsonString()
		if not fs.dir(self.hooksDir) then return end
		for hook in fs.dir(self.hooksDir) do
			local cStdin = io.popen(self.hooksDir.."/"..hook, "w")
			cStdin:write(jsonString)
			cStdin:close()
		end
	end
end

function SharedStateBase:save()
	if self.changed then
		local outFd = io.open(self.dataFile, "w")
		outFd:write(self:toJsonString())
		outFd:close()
		outFd = nil
	end
end

function SharedStateBase:httpRequest(url, body)
	local tmpfname = os.tmpname()

	local tmpfd = io.open(tmpfname, "w")
	tmpfd:write(body)
	tmpfd:close()
	tmpfd = nil

	local cmd = "uclient-fetch --no-check-certificate -q -O- --timeout=3 "
	cmd = cmd.."--post-file='"..tmpfname.."' '"..url.."' ; "
	cmd = cmd.."rm -f '"..tmpfname.."'"
	local fd = io.popen(cmd)

	local value = fd:read("*a")
	fd:close()

	return value
end

function SharedStateBase:_sync(urls)
	urls = urls or {}

	if #urls < 1 then
		local uci_cursor = uci:cursor()
		local fixed_candidates =
				uci_cursor:get("shared-state", "options","candidates") or {}
		for _, line in pairs(fixed_candidates) do
			table.insert(
				urls,
				line.."/"..self.dataType )
		end

		io.input(io.popen(shared_state.CANDIDATE_NEIGHBORS_BIN))
		for line in io.lines() do
			table.insert(
				urls,
				self:getSyncUrl(line, self.dataType))
		end
	end

	for _,url in ipairs(urls) do
		local body = self:toJsonString()

		local response = self:httpRequest(url, body)

		if type(response) == "string" and response:len() > 1  then
			local parsedJson = JSON.parse(response)
			if parsedJson then self:_merge(parsedJson) end
		else
			self.log( "debug", "error requesting "..url )
		end
	end
end

function SharedStateBase:sync(urls)
	self:lock()
	self:load()
	self:unlock()
	self:_sync(urls)
	self:lock()
	self:load(true) -- Take in account changes happened during sync
	self:save()
	self:unlock()
	self:notifyHooks()
end

function SharedStateBase:toJsonString()
	return JSON.stringify(self.storage)
end
{'fraction_non_alphanumeric': 0.1289620130655698, 'fraction_numerical': 0.004597144931042826, 'mean_word_length': 4.1798245614035086, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3629320', 'n_tokens_mistral': 3085, 'n_tokens_neox': 2873, 'n_words': 773}