Changed DHT storage to the newest patched MapDB (3.0.7) from my repo: https://github.com/ChronosX88/mapdb/

This commit is contained in:
ChronosX88 2019-04-27 18:06:23 +04:00
parent 924a951ba3
commit 686dc82d8c
9 changed files with 412 additions and 38 deletions

View File

@ -1,29 +0,0 @@
<component name="ProjectCodeStyleConfiguration">
<code_scheme name="Project" version="173">
<Objective-C-extensions>
<file>
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Import" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Macro" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Typedef" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Enum" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Constant" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Global" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Struct" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="FunctionPredecl" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Function" />
</file>
<class>
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Property" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="Synthesize" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="InitMethod" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="StaticMethod" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="InstanceMethod" />
<option name="com.jetbrains.cidr.lang.util.OCDeclarationKind" value="DeallocMethod" />
</class>
<extensions>
<pair source="cpp" header="h" fileNamingConvention="NONE" />
<pair source="c" header="h" fileNamingConvention="NONE" />
</extensions>
</Objective-C-extensions>
</code_scheme>
</component>

View File

@ -44,7 +44,9 @@ dependencies {
implementation "androidx.room:room-runtime:2.1.0-alpha04"
annotationProcessor "androidx.room:room-compiler:2.1.0-alpha04"
implementation 'org.slf4j:slf4j-log4j12:1.7.26'
implementation 'net.tomp2p:tomp2p-all:5.0-Beta8'
implementation('net.tomp2p:tomp2p-all:5.0-Beta8') {
exclude group: 'net.tomp2p', module: 'tomp2p-storage'
}
implementation 'com.google.android.material:material:1.1.0-alpha04'
implementation 'androidx.preference:preference:1.1.0-alpha03'
implementation 'com.google.code.gson:gson:2.8.5'
@ -54,7 +56,19 @@ dependencies {
implementation "org.jetbrains.anko:anko:0.10.8"
implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.0'
implementation 'com.esotericsoftware:kryo:5.0.0-RC1'
implementation 'org.eclipse.collections:eclipse-collections-api:7.0.0'
implementation 'org.eclipse.collections:eclipse-collections:7.0.0'
implementation 'org.eclipse.collections:eclipse-collections-forkjoin:7.0.0'
implementation('com.google.guava:guava:15.0') {
exclude group: 'com.google.guava', module: 'listenablefuture'
}
implementation 'net.jpountz.lz4:lz4:1.3.0'
implementation 'org.mapdb:elsa:3.0.0-M5'
}
repositories {
mavenCentral()
}
configurations {
all*.exclude group: 'com.google.guava', module:'listenablefuture'
}

Binary file not shown.

BIN
app/libs/mapdb-3.0.7.jar Normal file

Binary file not shown.

View File

@ -0,0 +1,109 @@
/*
* Copyright (C) 2019 ChronosX88
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.chronosx88.influence.helpers
import io.netty.buffer.Unpooled
import net.tomp2p.connection.SignatureFactory
import net.tomp2p.p2p.PeerBuilder
import net.tomp2p.storage.Data
import org.mapdb.DataInput2
import org.mapdb.DataOutput2
import org.mapdb.serializer.GroupSerializerObjectArray
import java.io.*
import java.nio.ByteBuffer
import java.security.InvalidKeyException
import java.security.SignatureException
class DataSerializerMapDB(private val signatureFactory: SignatureFactory) : GroupSerializerObjectArray<Data>() {
private val LOG_TAG = "DataSerializerMapDB"
override fun serialize(out: DataOutput2, value: Data) {
val dataOut = DataOutputStream(out)
val acb = Unpooled.buffer()
// store data to disk
// header first
if(value.publicKey().equals(PeerBuilder.EMPTY_PUBLIC_KEY)) {
value.publicKey(null)
}
value.encodeHeader(acb, signatureFactory)
writeData(dataOut, acb.nioBuffers())
acb.skipBytes(acb.writerIndex())
// next data - no need to copy to another buffer, just take the data
// from memory
writeData(dataOut, value.toByteBuffers())
// rest
try {
value.encodeDone(acb, signatureFactory)
writeData(dataOut, acb.nioBuffers())
} catch (e: InvalidKeyException) {
throw IOException(e)
} catch (e: SignatureException) {
throw IOException(e)
}
}
override fun deserialize(input: DataInput2, available: Int): Data {
val dataInput = DataInputStream(DataInput2.DataInputToStream(input))
var buf = Unpooled.buffer()
var data: Data? = null
while (data == null) {
buf.writeByte(dataInput.readByte().toInt())
data = Data.decodeHeader(buf, signatureFactory)
}
val len = data.length()
var me = ByteArray(len)
dataInput.readFully(me)
buf = Unpooled.wrappedBuffer(me)
var retVal = data.decodeBuffer(buf)
if (!retVal) {
throw IOException("data could not be read")
}
if (data.isSigned) {
me = ByteArray(signatureFactory.signatureSize())
dataInput.readFully(me)
buf = Unpooled.wrappedBuffer(me)
}
retVal = data.decodeDone(buf, signatureFactory)
if (!retVal) {
throw IOException("signature could not be read")
}
return data
}
@Throws(IOException::class)
private fun writeData(out: OutputStream, nioBuffers: Array<ByteBuffer>) {
val length = nioBuffers.size
for (i in 0 until length) {
val remaining = nioBuffers[i].remaining()
if (nioBuffers[i].hasArray()) {
out.write(nioBuffers[i].array(), nioBuffers[i].arrayOffset(), remaining)
} else {
val me = ByteArray(remaining)
nioBuffers[i].get(me)
out.write(me)
}
}
}
companion object {
private const val serialVersionUID = 1428836065493792295L
}
}

View File

@ -0,0 +1,278 @@
/*
* Copyright (C) 2019 ChronosX88
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.chronosx88.influence.helpers
import com.esotericsoftware.kryo.serializers.JavaSerializer
import net.tomp2p.connection.SignatureFactory
import net.tomp2p.dht.Storage
import net.tomp2p.peers.Number160
import net.tomp2p.peers.Number320
import net.tomp2p.peers.Number480
import net.tomp2p.peers.Number640
import net.tomp2p.storage.Data
import org.mapdb.BTreeMap
import org.mapdb.DB
import org.mapdb.DBMaker
import org.mapdb.serializer.SerializerJava
import java.io.File
import java.security.PublicKey
import java.util.*
import java.util.concurrent.ConcurrentHashMap
import kotlin.collections.ArrayList
class StorageMapDB(peerId: Number160, path : File, signatureFactory: SignatureFactory) : Storage {
// Core
private val dataMap: BTreeMap<Number640, Data>
// Maintenance
private val timeoutMap: BTreeMap<Number640, Long>
private val timeoutMapRev: BTreeMap<Long, Set<Number640>>
// Protection
private val protectedDomainMap: BTreeMap<Number320, PublicKey>
private val protectedEntryMap: BTreeMap<Number480, PublicKey>
// Responsibility
private val responsibilityMap: BTreeMap<Number160, Number160>
private val responsibilityMapRev: BTreeMap<Number160, Set<Number160>>
private val db: DB = DBMaker.fileDB(path).closeOnJvmShutdown().make()
private val storageCheckIntervalMillis: Int = 60 * 1000
init {
val dataSerializer = DataSerializerMapDB(signatureFactory)
//val kryoSerializer = KryoSerializer() // We use this because Elsa serializer works very poor
val javaSerializer = SerializerJava()
dataMap = db
.treeMap("dataMap_$peerId")
.valueSerializer(dataSerializer)
.createOrOpen() as BTreeMap<Number640, Data>
timeoutMap = db
.treeMap("timeoutMap_$peerId")
.createOrOpen() as BTreeMap<Number640, Long>
timeoutMapRev = db
.treeMap("timeoutMapRev_$peerId")
.createOrOpen() as BTreeMap<Long, Set<Number640>>
protectedDomainMap = db
.treeMap("protectedDomainMap_$peerId")
.createOrOpen() as BTreeMap<Number320, PublicKey>
protectedEntryMap = db
.treeMap("protectedEntryMap_$peerId")
.createOrOpen() as BTreeMap<Number480, PublicKey>
responsibilityMap = db
.treeMap("responsibilityMap_$peerId")
.createOrOpen() as BTreeMap<Number160, Number160>
responsibilityMapRev = db
.treeMap("responsibilityMapRev_$peerId")
.createOrOpen() as BTreeMap<Number160, Set<Number160>>
}
override fun contains(key: Number640?): Boolean {
return dataMap.containsKey(key)
}
override fun contains(from: Number640?, to: Number640?): Int {
return dataMap.subMap(from, true, to, true).size
}
override fun findContentForResponsiblePeerID(peerID: Number160?): MutableSet<Number160>? {
return responsibilityMapRev[peerID] as MutableSet<Number160>?
}
override fun findPeerIDsForResponsibleContent(locationKey: Number160?): Number160? {
return responsibilityMap[locationKey]
}
override fun put(key: Number640?, value: Data?): Data? {
val oldData = dataMap.put(key, value)
db.commit()
return oldData
}
override fun get(key: Number640?): Data? {
return dataMap[key]
}
override fun remove(key: Number640?, returnData: Boolean): Data? {
val retVal = dataMap.remove(key)
db.commit()
return retVal
}
override fun remove(from: Number640?, to: Number640?): NavigableMap<Number640, Data> {
val tmp = dataMap.subMap(from, true, to, true)
val retVal = TreeMap<Number640, Data>()
for(entry : Map.Entry<Number640, Data> in tmp.entries) {
retVal[entry.key] = entry.value
}
tmp.clear()
db.commit()
return retVal
}
override fun addTimeout(key: Number640, expiration: Long) {
val oldExpiration = timeoutMap.put(key, expiration)
putIfAbsent2(expiration, key)
if (oldExpiration == null) {
return
}
removeRevTimeout(key, oldExpiration)
db.commit()
}
private fun putIfAbsent2(expiration: Long, key: Number640) {
var timeouts = timeoutMapRev[expiration]
//var timeouts : MutableSet<Number640> = timeoutMapRev[expiration] as MutableSet<Number640>
if (timeouts == null) {
timeouts = Collections.newSetFromMap(ConcurrentHashMap())
}
(timeouts as MutableSet).add(key)
timeoutMapRev[expiration] = timeouts
}
private fun removeRevTimeout(key: Number640, expiration: Long?) {
val tmp = timeoutMapRev[expiration] as MutableSet<Number640>?
if (tmp != null) {
tmp.remove(key)
if (tmp.isEmpty()) {
timeoutMapRev.remove(expiration)
} else {
timeoutMapRev[expiration!!] = tmp
}
}
}
override fun updateResponsibilities(locationKey: Number160, peerId: Number160?): Boolean {
val oldPeerID = responsibilityMap.put(locationKey, peerId)
val hasChanged: Boolean
if (oldPeerID != null) {
if (oldPeerID == peerId) {
hasChanged = false
} else {
removeRevResponsibility(oldPeerID, locationKey)
hasChanged = true
}
} else {
hasChanged = true
}
var contentIDs: MutableSet<Number160>? = responsibilityMapRev[peerId] as MutableSet
if (contentIDs == null) {
contentIDs = HashSet()
}
contentIDs.add(locationKey)
responsibilityMapRev[peerId] = contentIDs
db.commit()
return hasChanged
}
private fun removeRevResponsibility(peerId: Number160, locationKey: Number160) {
val contentIDs = responsibilityMapRev[peerId] as MutableSet?
if (contentIDs != null) {
contentIDs.remove(locationKey)
if (contentIDs.isEmpty()) {
responsibilityMapRev.remove(peerId)
} else {
responsibilityMapRev[peerId] = contentIDs
}
}
}
override fun protectDomain(key: Number320?, publicKey: PublicKey?): Boolean {
protectedDomainMap[key] = publicKey
db.commit()
return true
}
override fun storageCheckIntervalMillis(): Int {
return storageCheckIntervalMillis
}
override fun isDomainProtectedByOthers(key: Number320?, publicKey: PublicKey?): Boolean {
val other = protectedDomainMap[key] ?: return false
return other != publicKey
}
override fun removeTimeout(key: Number640) {
val expiration = timeoutMap.remove(key) ?: return
removeRevTimeout(key, expiration)
db.commit()
}
override fun removeResponsibility(locationKey: Number160) {
val peerId = responsibilityMap.remove(locationKey)
if (peerId != null) {
removeRevResponsibility(peerId, locationKey)
}
db.commit()
}
override fun protectEntry(key: Number480?, publicKey: PublicKey?): Boolean {
publicKey ?: return true
protectedEntryMap[key] = publicKey
return true
}
override fun map(): NavigableMap<Number640, Data> {
val retVal = TreeMap<Number640, Data>()
for ((key, value) in dataMap) {
retVal[key] = value
}
return retVal
}
override fun isEntryProtectedByOthers(key: Number480?, publicKey: PublicKey?): Boolean {
val other = protectedEntryMap[key] ?: return false
return other != publicKey
}
override fun subMap(from: Number640?, to: Number640?, limit: Int, ascending: Boolean): NavigableMap<Number640, Data> {
val tmp = dataMap.subMap(from, true, to, true)
val retVal = TreeMap<Number640, Data>()
if (limit < 0) {
for ((key, value) in if (ascending) tmp else tmp.descendingMap()) {
retVal[key] = value
}
} else {
val limit1 = Math.min(limit, tmp.size)
val iterator = if (ascending)
tmp.entries.iterator()
else
tmp.descendingMap().entries.iterator()
var i = 0
while (iterator.hasNext() && i < limit1) {
val entry = iterator.next()
retVal[entry.key] = entry.value
i++
}
}
return retVal
}
override fun subMapTimeout(to: Long): MutableCollection<Number640> {
val tmp = timeoutMapRev.subMap(0L, to)
val toRemove = ArrayList<Number640>()
for (set in tmp.values) {
toRemove.addAll(set)
}
return toRemove
}
override fun close() {
db.close()
}
}

View File

@ -14,6 +14,7 @@ import net.tomp2p.connection.Ports;
import net.tomp2p.connection.RSASignatureFactory;
import net.tomp2p.dht.PeerBuilderDHT;
import net.tomp2p.dht.PeerDHT;
import net.tomp2p.dht.Storage;
import net.tomp2p.futures.FutureBootstrap;
import net.tomp2p.futures.FutureDiscover;
import net.tomp2p.nat.FutureRelayNAT;
@ -26,6 +27,7 @@ import net.tomp2p.relay.tcp.TCPRelayClientConfig;
import net.tomp2p.replication.IndirectReplication;
import net.tomp2p.storage.Data;
import java.io.File;
import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
@ -42,7 +44,7 @@ import io.github.chronosx88.influence.helpers.JVMShutdownHook;
import io.github.chronosx88.influence.helpers.KeyPairManager;
import io.github.chronosx88.influence.helpers.NetworkHandler;
import io.github.chronosx88.influence.helpers.P2PUtils;
import io.github.chronosx88.influence.helpers.StorageBerkeleyDB;
import io.github.chronosx88.influence.helpers.StorageMapDB;
import io.github.chronosx88.influence.helpers.actions.UIActions;
import io.github.chronosx88.influence.models.PublicUserProfile;
@ -59,7 +61,7 @@ public class MainLogic implements CoreContracts.IMainLogicContract {
private IndirectReplication replication;
private KeyPairManager keyPairManager;
private Thread checkNewChatsThread = null;
private StorageBerkeleyDB storage;
private Storage storage;
public MainLogic() {
this.context = AppHelper.getContext();
@ -83,8 +85,9 @@ public class MainLogic implements CoreContracts.IMainLogicContract {
new Thread(() -> {
try {
StorageBerkeleyDB storageBerkeleyDB = new StorageBerkeleyDB(peerID, context.getFilesDir(), new RSASignatureFactory());
this.storage = storageBerkeleyDB;
//StorageBerkeleyDB storageBerkeleyDB = new StorageBerkeleyDB(peerID, context.getFilesDir(), new RSASignatureFactory());
Storage storageMapDB = new StorageMapDB(peerID, new File(context.getFilesDir(), "dhtDB.db"), new RSASignatureFactory());
this.storage = storageMapDB;
peerDHT = new PeerBuilderDHT(
new PeerBuilder(peerID)
.ports(7243)
@ -92,9 +95,9 @@ public class MainLogic implements CoreContracts.IMainLogicContract {
.channelServerConfiguration(createChannelServerConfig())
.start()
)
.storage(storageBerkeleyDB)
.storage(storageMapDB)
.start();
Runtime.getRuntime().addShutdownHook(new JVMShutdownHook(storageBerkeleyDB));
Runtime.getRuntime().addShutdownHook(new JVMShutdownHook(storageMapDB));
try {
String bootstrapIP = this.preferences.getString("bootstrapAddress", null);
if(bootstrapIP == null) {

View File

@ -40,7 +40,7 @@ class SettingsLogic : CoreContracts.ISettingsLogic {
data!!.protectEntry(mainKeyPair)
val isSuccess = P2PUtils.put(username, null, data, mainKeyPair)
Log.i(LOG_TAG, if (isSuccess) "Username $username is published!" else "Username $username isn't published!")
Log.i(LOG_TAG, if (isSuccess) "# Username $username is published!" else "# Username $username isn't published!")
} ?: run {
return
}

View File

@ -16,4 +16,3 @@ android.useAndroidX=true
android.enableJetifier=true
android.enableD8=true