最近我在Github上发现了一个新推出的项目 - Khipu，它涉及对以太坊的两项改进：1）并行执行事务2）存储性能优化。这两点也是Dora网络的技术特征和优势的一部分。很高兴看到更多项目优化区块链性能。本文介绍了Khipu项目的内容，实现方法和当前状态。

首先看一下Kipu官方github的介绍（来自https://github.com/khipu-io）

也就是说，Khipu是一个基于Scala/Akka的以太坊协议，目前正在开发中。已经发布的alpha版本的主要功能包括：

1.尽可能并行执行块内的事务。目前，块的内部事务的80％可以并行执行。
2.为区块链设计并实施了一个存储引擎（Kesque）。基于Kafka的日志引擎开发，超过99％的随机读取只需要一个磁盘io。

我们来看看项目的进展情况

Khipu目前实施了3个大型模块：

*节点发现（访问）
*快速同步（同步到最近的状态数据快照和所有块）
*定期同步（同步块和执行块中包含的事务）

稍后将要开发的功能将不再详述。简而言之，当前的0.1.0-alpha版本实现了一个不完整的节点（不是轻节点，因为它不是光同步块头）或数据备份节点。它可以同步块，执行和验证块事务的合法性。但它不能作为一个完整的节点存在，因为它不能被接受，或者它可以生成交易，并且由于挖掘共识尚未实现而无法发布。虽然协议的实施仍未完成，但在项目的设计和实施中有许多值得参考的地方。还可以看出，开发人员已经做了一些深入的思考。

Khipu的设计理念

与github上的介绍一致，Khipu的设计理念相对清晰，如图:所示

整个服务由两个大块组成：Akka部分主要负责处理块（内部事务），Kafka部分完成数据库功能，用于存储块和各种状态数据。首先让我们看看Kipuu如何处理块。

块处理

由于Kipu实现基于Akka，让我们简要介绍一下akka和actor模型。

什么是akka

Akka是一个用scala编写的库，用于简化容错，可伸缩和高并发应用程序的编写。 Akka使用actor模型来增强抽象，并为构建可伸缩，灵活的应用程序提供更好的平台。对于更难的错误，akka使用“让它崩溃”模型。此模式可以使任务的处理失败不会导致整个应用程序崩溃，从而使您的系统具有强大的自我修复功能，并且无需重新启动即可恢复系统。同时，akka的分布式部署更简单，更透明。

演员模特

演员模型不是新业务。它早在20世纪70年代就已提出，其主要目的是解决分布式编程中的一系列问题。 actor模型具有以下优点：1。更简单，高度抽象的并发处理。 2.异步，非阻塞，高性能事件驱动编程模型。 3.非常轻量级的事件驱动处理。

基于以上两个模块，Kipu的块处理逻辑并不复杂。主要工作是并行执行块事务。这种并行处理的逻辑是一组map reduce。 （极端）简化过程如下:(感谢Scala的map reduce功能，Java用于多次估算代码量）

这里，Map任务很容易理解。单独块中的事务在多个并行世界中执行以执行它们各自的事务并改变它们各自的状态。主要问题是Reduce阶段，如何合并该系列的世界状态。在合并这篇文章时，Khipu采用了冲突检测方法，定义了两种不同的冲突模式（并行状态冲突和世界状态冲突）。

对象ProgramState {
特质ParallelRace
案例对象OnAccount扩展了ParallelRace
Case对象OnError扩展了ParallelRace
}

Object BlockWorldState {
密封性状RaceCondition
Case对象OnAddress扩展了RaceCondition
案例对象OnAccount扩展了RaceCondition
案例对象OnStorage扩展了RaceCondition
案例对象OnCode扩展了RaceCondition
}
冲突检测的基本逻辑非常清楚。例如，如果两个世界状态访问同一个帐户，肯定存在冲突，或者如果在并行执行期间发生错误，则很可能是由并行执行引起的。无论检测到何种冲突，结果都与序列化执行模式的回退相同，即串行执行的顺序与以太坊块包事务的顺序相同。个人认为在这里定义冲突没有问题，但冲突类型的定义和检测不够全面，但考虑到alpha版本只是一个数据节点，后续的必须加强。

数据存储

Khipu设计了Kesque，一种基于Kafka Log引擎的数据读/写引擎，试图以类似于流消息队列的方式重新定义区块链数据存储。

简要介绍卡夫卡

Apache KafkaTM是一个分布式流媒体平台，通常被认为具有三个关键功能：流媒体平台：

1.它允许您发布和订阅记录流。从这个角度来看，它接近于消息队列或企业消息传递系统。
2.它允许您以容错方式存储流记录。
3.它允许您即时处理消息流。

Kafka的优势体现在两个主要应用中：1。构建实时流数据流水线，以便可靠，一致地在应用系统之间获取数据。 2.构建实时流应用程序，响应数据流。

Kafka工作模式：1。Kafka在群集中的一个或多个服务器上运行。 2. Kafka存储流记录按主题分类。 3.每条记录包含一个键，一个值和一个时间戳。

为什么Khapu选择Kafka

Khipu对Kafka的采用应该基于其持久化流数据的能力以及它在持久实现（主题/索引等）中的组织方式。这件作品和区块链应用程序也非常接近。对于以太坊上的数据，相当一部分数据，特别是MPT和账户相关数据是额外的写随机读数，也就是说，仅使用加法和删除。说到曾和，它不会被改变或删除。需要存储许多类型的数据。因此，Khipu定义了许多主题表来区分不同的数据，也可以将其视为存储并行化的一种手段。创建主题表的（部分）代码如下：
Lazy val kesque=new Kesque（kafkaProps）
Log.info（s'Kesque开始使用配置文件: $ kafkaConfigFile'）
私有val futureTables=Future.sequence（列表（
Future（kesque.getTable（Array（KesqueDataSource.account））），
Future（kesque.getTable（Array（KesqueDataSource.storage））），
Future（kesque.getTable（Array（KesqueDataSource.evmcode））），
未来（kesque.getTimedTable（数组）（
KesqueDataSource.header，
KesqueDataSource.body，
KesqueDataSource.receipts
），1024000））
））
私有val列表（accountTable，storageTable，evmcodeTable，blockTable）=Await.result（futureTables，Duration.Inf）
//private val headerTable=kesque.getTimedTable（Array（KesqueDataSource.header），1024000）
//private val bodyTable=kesque.getTable（Array（KesqueDataSource.body），1024000）
//private val receiptTable=kesque.getTable（Array（KesqueDataSource.receipts），1024000）

Lazy val accountNodeDataSource=new KesqueDataSource（accountTable，KesqueDataSource.account）
Lazy valNodeDataSource=new KesqueDataSource（storageTable，KesqueDataSource.storage）
Lazy val evmCodeDataSource=new KesqueDataSource（evmcodeTable，KesqueDataSource.evmcode）

Lazy valHeadersDataSource=new KesqueDataSource（blockTable，KesqueDataSource.header）
Lazy valBodiesDataSource=new KesqueDataSource（blockTable，KesqueDataSource.body）
Lazy valtstsDataSource=new KesqueDataSource（blockTable，KesqueDataSource.receipts）
我在上面看到的是Kipu用于存储KesqueDataSource的一些主题表。您可以看到数据主要与帐户相关，其余数据仍存储在levelDB中。

值得一提的是，Khipu随机读取了一个索引偏移hashOffsets映射表。通过该表，您可以直接访问日志文件中的相应值，如果不是直接从Cache中获取的话。个人认为这就是为什么它声称99％随机阅读只需要一个IO。阅读部分的代码如下：
Def读取（键:数组[字节]，主题:字符串）:选项[TVal]={ 试试{
readLock.lock

Val valueIndex=topicIndex（topic）
缓存（valueIndex）.get（哈希（键））匹配{
案例无=>
Val hash=Hash（key）
hashOffsets.get（hash.hashCode，valueIndex）匹配{
Case IntIntsMap.NO_VALUE=>无
案例抵消=>
Var foundValue:选项[TVal]=无
Var foundOffset=Int.MinValue
Var i=offsets.length - 1 //向后循环以找到最新的
而（i>=0＆amp;＆amp; foundValue.isEmpty）{ Val offset=偏移量（i）
Val（topicPartition，result）=db.read（topic，offset，fetchMaxBytes）.head
Val recs=result.info.records.records.iterator
while（recs.hasNext）{//注意:记录被抵消了!!
Val rec=recs.next
if（rec.offset==offset＆amp;＆amp; java.util.Arrays.equals（db.getBytes（rec.key），key））{ foundOffset=offset
foundValue=if（rec.hasValue）some（TVal（db.getBytes（rec.value），rec.timestamp））else else
}
}
我 - =1
}

foundValue foreach {x=>
缓存（valueIndex）.put（hash，（x，foundOffset））
}

foundValue
}
Case Some（（value，offset））=> Some（value）
}
终于{ readLock.unlock（）
}
}
这里有一个小问题。我觉得这个hashOffsets表可能会给内存带来一些压力，可能是因为当前的数据量不够大，所以Khipu似乎没有考虑表本身可能需要溢出到磁盘的问题。

简而言之，Khapu在数据存储方向上的尝试非常值得认可，尽管目前以太网实施（go版）的性能和空间水平可以满足要求，但随着区块链技术的进一步发展，不可避免地要提高存储要求，区块链技术和大数据（至少存储方面）框架集成应该是一个非常有前途的方向。

总结

Khipu是一个用Scala语言实现的以太坊数据节点协议，结合了Scala语言特性，Akka，Kafka和其他框架。在事务执行和数据存储方面已经进行了一些有意义的尝试。代码中包含的设计和思想，包括Scala语言，Akka/Kafka框架的选择对于基于以太坊的其他项目也具有指导意义。