首页 文章 精选 留言 我的

精选列表

搜索[源码阅读],共10000篇文章
优秀的个人博客,低调大师

Giraph源码分析(六)——Edge 分析

1.在Vertex类中,顶点的存储方式采用邻接表形式。每个顶点有 VertexId、VertexValue、OutgoingEdges和Halt,boolean型的halt变量用于记录顶点的状态,false时表示active,true表示inactive状态。 片段代码如下。 2.org.apache.giraph.edge.Edge 接口,用于存储顶点的边,每条边包含targetVertexId和edgeValue两个属性。类关系图如下: Giraph默认使用DefaultEdge类存储边,该类中有两个变量: I targetVertexId和 E value,I为顶点ID的类型,E为边的类型。注意,DefaultEdge类同时继承ReusableEdge接口,在ReusableEdge类的定义中,有如下说明文字:A complete edge, the target vertex and the edge value. Can only be one edge with a destination vertex id per edge map. This edge can be reused, that is you can set it's target vertex ID and edge value. Note: this class is useful for certain optimizations, but it's not meant to be exposed to the user. Look at MutableEdge instead. 从上述说明文字可知,edge可以被重用,只需要修改targetVertexId和value的值就行。即每个Vertex若有多条出边,只会创建一个DefaultEdge对象来存储边。3.org.apache.giraph.edge.OutEdges 用于存储每个顶点的out-edges。从Vertex类的定义可知,顶点的每条边都被存储在OutEdges类型的edge对象中,OutEdges接口的关系图如下: Giraph默认的使用ByteArrayEdges,每个顶点的所有边都被存储在byte[ ]中。当顶点向它的出边发送消息时,需要遍历Vertex类中的edges对象。示例代码如下: 注意:由DefaultEdge的定义可知,遍历getEdges时,返回的Edge对象时同一个对象,只是该对象中值改变了。下面继续查看代码来证明此观点。查看ByteArrayEdges类的iterator()方法,如下: 返回的是内部类ByteArrayEdgeIterator对象,定义如下: 总结:当顶点的出度很大时,此优化甚好,能很好的节约内存。如UK-2005数据中,顶点的最大出度为 5213。假设顶点1的出度顶点有<2 , 0.4>,<3 , 7.8> ,<5 , 6.4> 。如下代码: 输出结果为:[ 2 ][ 3 , 3 ][ 5 , 5 , 5 ]并非是希望的 [ 2 , 3 , 5 ]

优秀的个人博客,低调大师

Giraph源码分析(三)—— 消息通信

由前文知道每个BSPServiceWorker有一个WorkerServer对象,WorkerServer对象里面又有ServerData对象,作为数据实。ServerData中包含该Worker的partitionStore、edgeStore、incomingMessageStore、currentMessageStore、聚集值等。其中incomingMessageStore对象为MessageStoreByPartition(接口)类型,也就是说消息时按照分区来存储的。MessageStoreByPartition接口的关系图如下: 在SimpleMessageStore抽象类中,有一个ConcurrentMap>类型的变量map,用来存储消息。第一层是pairtitionID到发送到该partition消息的映射;第二层是VertexID 到发送给该Vertex的消息队列。 《Giraph通信模块分析》:http://my.oschina.net/skyaugust/blog/95182 每个顶点的消息列表具体为ExtendedDataOutput类型,它继承DataOutput接口,增加了几个方法而已。每个消息是以字节形式写入到ExtendedDataOutput对象中的。 发送消息时,采用异步式通信。 图顶点的计算处理与消息通信并发执行,在计算过程中就可以发送消息,将大规模消息发送分散在不同的时间段,避免瞬时网络通信阻塞,但是接受端需要额外的空间,存储临时接收到的消息,相当于空间换时间。而集中式通信,图顶点的计算处理与消息通信串行进行,在计算完毕后,统一发送消息,控制和实现方式简单,可在发送端对消息进行最大程度优化,但容易造成瞬时间的网络通信阻塞以及增加发送端的消息存储开销。 不同Worker间的消息通信使用RPC方式,具体为Netty。同一Worker内,连续两次迭代的消息直接通过内存操作,把要发送的消息直接复制到Worker的incomingMessageStore中。下面详述消息的存储格式和发送机制。 Giraph使用Cache来缓存消息,当消息达到一定阈值后,一次性发送。 既按照bulk模式进行,不会一条一条信息发送。向某个顶点发送的消息是按照 pair存储在ByteArrayVertexIdData中(实际为ByteArrayVertexIdMessages类型)。介绍如下: org.apache.giraph.utils.ByteArrayVertexIdData 功能:把<顶点ID,data> Pair 存储在一个 byte数组中。里面有 ExtendedDataOutput对象用来存储数据。 该类中还有一个内部类:VertexIdDataIterator,该内部类继承 VertexIdIterator类。 org.apache.giraph.comm.SendCache用来缓存发送的信息,然后以“Bulk”模式发送。在Giraph中,每个Worker上可以对应多个分区。消息缓存的阈值是以Worker为单位计算,而不是Partition。 SendCache中有ByteArrayVertexIdData[ ] dataCache数组用来存储发送给每个Partition的消息;有int[ ] dataSizes数组用于记录向每个Worker发送的消息大小,若大于MAX_MSG_REQUEST_SIZE(默认为512KB)就把此Worker上的所有Partition缓存的消息发送到给该Worker,同一Worker内消息也是如此缓存;有int[ ] initBufferSizes数组用于记录每个Worker上的每个Partition的初始化ByteArrayVertexIdData中ExtendedDataOutput对象的大小,同一Worker上的所有Partition初始值相同,该值为平均值。记MAX_MSG_REQUEST_SIZE(message request size)值为M, 该Worker上有P个 partitions,ADDTITIONNAL_MSG_REQUEST_SIZE(比平均值大的因子)默认为0.2f,记为A。则每个Partition的初始大小为:M*(1+A) / P . 由前文知道,每个Worker都有一个NettyWorkerClientRequestProcessor用来发送消息。该类中有SendMessageCache对象用来缓存向外发送的信息。NettyWorkerClientRequestProcessor类中的sendMessageRequest(I,M) 方法如下,用于向某个顶点destVertexId发送消息message。 方法解释:首先根据destVertexId得到对应的partitionId和WorkerInfo,然后把消息add到SendMessageCache中,并返回向该顶点所属Worker发送的消息大小workerMessageSize。若该值大于默认值512KB,则把此Worker对应的所有Partition消息从SendMessageCache中删除,把删除的消息赋值给workerMessages,其类型为PairList> ,key为partitionId,value为发送给该partition的消息列表,最后调用doRequest()方法发送信息。doRequest()方法如下: 可以看到在发送消息时,先判断是否在同一Worker上。如果是的话,调用SendWorkerMessagesRequest的doRequest发送消息;否则使用WorkerClient(底层使用Netty)进行消息发送。下面着重讨论同一Worker内的机制。 org.apache.giraph.comm.requests.SendWorkerMessagesRequest类中的doRequest方法如下: 参数为该Worker的ServerData,代码中的partitionVertexData实际为PairList>workerMessages。遍历来添加到ServerData中的incomingMessageStore中。 ByteArrayMessagesPerVertexStore类中的addPartitionMessages()方法如下: 当用户使用了Combiner,incomingMessageStore对应的类型则为OneMessagePerVertexStore,该类为每个顶点只存储一个消息,而非消息队列。 结构如下图: 当添加一条消息时,会把顶点已对应的消息和要添加的消息调用combine()方法进行合并,然后存储在上述结构图中。addPartitionMessages()方法如下: 在ComputeCallable中的call()方法调用computePartition(Partition)计算完所有Partition上的顶点后,调用WorkerClientRequestProcessor.flush()方法把所有剩余的消息发送出去。

优秀的个人博客,低调大师

kubernetes controller源码解读之StatefulSet

1. StatefulSet应用场景说明 Deployment部署的无状态应用,应用的各个实例是相互独立的。但是在实际应用中存在如下需求: 应用的各个实例之间有一定依赖关系。如多个实例组成一个集群(比如ETCD集群),实例之间需要通过通信选出leader。这就要求各个实例的网络地址必须固定(不管是实例重启,还是热迁移到别的节点等实例的网络地址都必须保持不变),否则就无法组成稳定集群。 应用的实例和存储数据需要绑定。比如部署mysql集群(主备模式),两个实例通过 sharding来保存数据,所以访问这两个实例是分库之后不同的数据。而用户会要求访问两个实例返回数据是稳定的,也就是要求实例和存储数据必须要绑定(不管是实例是重启,或者迁移到其他节点),实例对应的 storage必须固定。 应用的各实例启动需要遵循一定的顺序。比如首先启动的实例必

优秀的个人博客,低调大师

kubernetes controller源码解读之DaemonSet

1. 适用场景 通过 DaemonSet部署的应用(Pod)主要用于满足如下场景: 类似守护进程,每个节点保证部署一个应用 能跟随节点的新增/移除,自动创建/删除守护应用 可以方便的对守护应用进行版本升级或者回滚 实际应用场景中,每个节点都需要的agent类型组件(如日志收集组件fluentd等),一般都采用DaemonSet方式部署。 2. DaemonSet资源定义 单个 DaemonSet资源的定义结构( DaemonSet的yam定义需要遵守该结构)如下: type DaemonSet struct { metav1.TypeMeta metav1.ObjectMeta Spec DaemonSetSpec Status DaemonSetStatus } DaemonSets Controller将根据 D

优秀的个人博客,低调大师

iOS 短视频源码开发MPMoviePlayerController

文:布谷惠泽/来源:山东布谷鸟网络 MPMoviePlayerController用来播放视频,在iOS9之后被弃用(iOS9之后苹果推荐我们使用AVPlayer,AVPlayer相对复杂但灵活),由于APP往往要兼容iOS9之前的版本,所有MPMoviePlayerController还是很重要的。 在我的另一篇文章中分享了一个基于MPMoviePlayerController的播放器,大家可以看看,目前还不完整。小伙伴们可以关注一下我的简书。谢谢 MPMoviePlayerController的简单使用 需要添加这个框架MediaPlayer.framework #import <MediaPlayer/MediaPlayer.h> #pragma mark - 本地 NSString* _moviePath=[[NSBundle mainBundle]pathForResource:@"popeye" ofType:@"mp4"]; self.player=[[MPMoviePlayerController alloc]initWithContentURL:[NSURL fileURLWithPath:_moviePath]]; [self.view addSubview:self.player.view]; self.player.view.frame=CGRectMake(0, 0, self.view.frame.size.width, CGRectGetWidth(self.view.frame)*(9.0/16.0)); self.player.movieSourceType = MPMovieSourceTypeFile;// 播放本地视频时需要这句 // self.player.controlStyle = MPMovieControlStyleNone;// 不需要进度条 self.player.shouldAutoplay = YES;// 是否自动播放(默认为YES) // self.player.scalingMode=MPMovieScalingModeAspectFill; [self.player prepareToPlay];//缓存 // [self.player play];//可以不加这句 #pragma mark - 网络 NSURL* url = [NSURL URLWithString:@"https://clips.vorwaerts-gmbh.de/big_buck_bunny.mp4"]; _player = [[MPMoviePlayerController alloc] initWithContentURL:url]; [self.view addSubview:self.player.view]; self.player.view.frame=CGRectMake(0, 0, self.view.frame.size.width, CGRectGetWidth(self.view.frame)*(9.0/16.0)); [self.player prepareToPlay]; [self.player play]; #pragma mark - 直播 NSURL* url = [NSURL URLWithString:@"http://live.hkstv.hk.lxdns.com/live/hks/playlist.m3u8"]; _player = [[MPMoviePlayerController alloc] initWithContentURL:url]; [self.view addSubview:self.player.view]; self.player.view.frame=CGRectMake(0, 0, self.view.frame.size.width, CGRectGetWidth(self.view.frame)*(9.0/16.0)); self.player.controlStyle=MPMovieSourceTypeStreaming;//直播 [self.player prepareToPlay]; // [self.player play]; MPMoviePlayerController提供了很多通知,这里我就简单的监听2个。我们可以通过监听到的信息做相应的处理。 #pragma mark - Notification //监听视频播放结束 [[NSNotificationCenter defaultCenter]addObserver:self selector:@selector(endPlay) name:MPMoviePlayerPlaybackDidFinishNotification object:nil]; //监听当前视频播放状态 [[NSNotificationCenter defaultCenter]addObserver:self selector:@selector(loadStateDidChange:) name:MPMoviePlayerLoadStateDidChangeNotification object:nil]; #pragma mark - Notification function -(void)endPlay { NSLog(@"播放结束"); } -(void)loadStateDidChange:(NSNotification*)sender { switch (self.player.loadState) { case MPMovieLoadStatePlayable: { NSLog(@"加载完成,可以播放"); } break; case MPMovieLoadStatePlaythroughOK: { NSLog(@"缓冲完成,可以连续播放"); } break; case MPMovieLoadStateStalled: { NSLog(@"缓冲中"); } break; case MPMovieLoadStateUnknown: { NSLog(@"未知状态"); } break; default: break; } } #pragma mark - dealloc - (void)dealloc { [[NSNotificationCenter defaultCenter]removeObserver:self]; }

优秀的个人博客,低调大师

LevelDB源码分析-sstable的Block

sstable中的Block(table/block.h table/block.cc table/block_builder.h table/block_builder.cc) sstable中的block由Block类封装并由BlockBuilder类构建。 block的结构为: entry1 | entry2 | ... | restarts(uint32 * num_of_restarts) | num_of_restarts(uint32) | trailer 其中entry格式为: shared_bytes(varint32) | unshared_bytes(varint32) | value_bytes(varint32) | unshared_key_data(unshared_bytes) | value_data(value_bytes) shared_bytes为这个entry中的key与前一个entry的key共享的字节数,entry的key在存储时只存储不共享的部分,即采用前缀压缩的形式,并且前缀压缩是分组进行的,以避免所有查找都需要从头开始的情况。而每个组的起始entry的位置(offset)存储在restarts中,每个为uint32,数量为num_of_restarts。 trailer的格式为: type(char) | crc(uint32) type为block的压缩方式,crc为循环冗余校验码。 Block类 Block类对sstable的block中的数据进行封装,并提供了一个迭代器作为接口。Block类的声明为: class Block { public: // ... private: // ... const char *data_; size_t size_; uint32_t restart_offset_; // Offset in data_ of restart array bool owned_; // Block owns data_[] // ... class Iter; }; 其中成员变量data_为指向block中数据的指针,size_为block中数据的大小,restart_offset_为block中restarts部分数据的起始位置的offset,owned标记block是否包含data_。 而LevelDB实际上为Block类封装了一个Iter类,对Block的遍历操作需要通过Iter类提供的接口函数进行。 class Block::Iter : public Iterator { private: const Comparator *const comparator_; const char *const data_; // underlying block contents uint32_t const restarts_; // Offset of restart array (list of fixed32) uint32_t const num_restarts_; // Number of uint32_t entries in restart array // current_ is offset in data_ of current entry. >= restarts_ if !Valid uint32_t current_; uint32_t restart_index_; // Index of restart block in which current_ falls std::string key_; Slice value_; Status status_; // ... public: // ... virtual bool Valid() const { return current_ < restarts_; } virtual Status status() const { return status_; } virtual Slice key() const { // ... } virtual Slice value() const { // ... } virtual void Next() { // ... } virtual void Prev() { // ... } virtual void Seek(const Slice &target) { // 见下文分析 } virtual void SeekToFirst() { // ... } virtual void SeekToLast() { // ... } private: // ... }; 其中,前几个成员变量的含义与Block类中的成员变量的含义类似,这里不再赘述。key_、value_这两个成员变量为迭代器当前指向的entry的KV值。 在这里主要分析Seek(const Slice &target)函数,其余函数的实现比较简单,在这里就不做分析了。Seek函数首先通过二分查找算法找到target所在的前缀编码分组的位置: // Binary search in restart array to find the last restart point // with a key < target uint32_t left = 0; uint32_t right = num_restarts_ - 1; while (left < right) { uint32_t mid = (left + right + 1) / 2; uint32_t region_offset = GetRestartPoint(mid); uint32_t shared, non_shared, value_length; const char *key_ptr = DecodeEntry(data_ + region_offset, data_ + restarts_, &shared, &non_shared, &value_length); if (key_ptr == nullptr || (shared != 0)) { CorruptionError(); return; } Slice mid_key(key_ptr, non_shared); if (Compare(mid_key, target) < 0) { // Key at "mid" is smaller than "target". Therefore all // blocks before "mid" are uninteresting. left = mid; } else { // Key at "mid" is >= "target". Therefore all blocks at or // after "mid" are uninteresting. right = mid - 1; } } 然后在该分组中遍历: // Linear search (within restart block) for first key >= target SeekToRestartPoint(left); while (true) { if (!ParseNextKey()) { return; } if (Compare(key_, target) >= 0) { return; } } 其中调用了DecodeEntry函数: static inline const char *DecodeEntry(const char *p, const char *limit, uint32_t *shared, uint32_t *non_shared, uint32_t *value_length) { // ... } 这个函数将以p为起始位置的entry解码,并且解码不会超过limit指向的位置,将key共享字节长度、key非共享字节长度、value长度存入shared,non_shared,value_length,返回指向entry中unshared_key_data起始位置的指针。 还调用了ParseNextKey函数: bool ParseNextKey() { // ... } 这个函数获取下一个entry的key和value值并存入迭代器的成员变量中。 BlockBuilder类 BlockBuilder类用于构建sstable中的block。BlockBuilder类声明为: class BlockBuilder { public: // ... // REQUIRES: Finish() has not been called since the last call to Reset(). // REQUIRES: key is larger than any previously added key void Add(const Slice &key, const Slice &value); // Finish building the block and return a slice that refers to the // block contents. The returned slice will remain valid for the // lifetime of this builder or until Reset() is called. Slice Finish(); // ... private: const Options *options_; std::string buffer_; // Destination buffer std::vector<uint32_t> restarts_; // Restart points int counter_; // Number of entries emitted since restart bool finished_; // Has Finish() been called? std::string last_key_; // ... }; 其中options_为构建选项,buffer_为实际block中的数据,restarts_存储前缀编码各个分组的起始位置,counter_记录当前前缀编码分组中已经压缩了几个entry,finished_标记Finish是否被调用,last_key_为block中上一次添加的key值。 下面我们首先看Add函数: void BlockBuilder::Add(const Slice &key, const Slice &value) Add函数首先检查当前前缀编码分组中压缩的entry个数是否已经达到上限,如果没有达到上限则计算共享的key的字节个数: Slice last_key_piece(last_key_); assert(!finished_); assert(counter_ <= options_->block_restart_interval); assert(buffer_.empty() // No values yet? || options_->comparator->Compare(key, last_key_piece) > 0); size_t shared = 0; if (counter_ < options_->block_restart_interval) { // See how much sharing to do with previous string const size_t min_length = std::min(last_key_piece.size(), key.size()); while ((shared < min_length) && (last_key_piece[shared] == key[shared])) { shared++; } } 如果达到上限则重新开始一个restart: else { // Restart compression restarts_.push_back(buffer_.size()); counter_ = 0; } const size_t non_shared = key.size() - shared; 然后依次将相应的数据编码后组成一个新的entry: // Add "<shared><non_shared><value_size>" to buffer_ PutVarint32(&buffer_, shared); PutVarint32(&buffer_, non_shared); PutVarint32(&buffer_, value.size()); // Add string delta to buffer_ followed by value buffer_.append(key.data() + shared, non_shared); buffer_.append(value.data(), value.size()); 最后将数据加在buffer_的后面并更新last_key_和counter_的值: // Update state last_key_.resize(shared); last_key_.append(key.data() + shared, non_shared); assert(Slice(last_key_) == key); counter_++; 然后是Finish函数: Slice BlockBuilder::Finish() { // Append restart array for (size_t i = 0; i < restarts_.size(); i++) { PutFixed32(&buffer_, restarts_[i]); } PutFixed32(&buffer_, restarts_.size()); finished_ = true; return Slice(buffer_); } Finish函数首先将restarts_的偏移量存入buffer_,然后存入num_of_restarts,然后将buffer_封装为一个Slice返回。 228 Love u

优秀的个人博客,低调大师

java源码-java.util.List

public List<String> findAddressesByApplication(String application) { List<String> ret = new ArrayList<String>(); ConcurrentMap<String, Map<Long, URL>> consumerUrls = getRegistryCache().get(Constants.CONSUMERS_CATEGORY); if(consumerUrls == null) return ret; for(Map.Entry<String, Map<Long, URL>> e1 : consumerUrls.entrySet()) { Map<Long, URL> value = e1.getValue(); for(Map.Entry<Long, URL> e2 : value.entrySet()) { URL u = e2.getValue(); if(application.equals(u.getParameter(Constants.APPLICATION_KEY))) { String addr = u.getAddress(); if(addr != null) ret.add(addr); } } } return ret; } 原文:http://www.codingdict.com/sources/java/java.util/5364.html

优秀的个人博客,低调大师

Java 源码解析实战 - ThreadLocal 原理

说起CS游戏,应该是每个中二少年的年少回忆了.游戏开始时,每个人能够领到一把枪,枪把上有三个数字:子弹数、杀敌数、自己的命数,为其设置的初始值分别为1500、0、10. 设战场上的每个人都是一个线程,那么这三个初始值写在哪里呢?如果每个线程都写死这三个值,万一将初始子弹数统一改成 1000发呢?如果共享,那么线程之间的并发修改会导致数据不准确.能不能构造这样一个对象,将这个对象设置为共享变量,统一设置初始值,但是每个线程对这个值的修改都是互相独立的.这个对象就是ThreadLocal 注意不能将其翻译为线程本地化或本地线程英语恰当的名称应该叫作:CopyValueIntoEveryThread 示例代码该示例中,无 set 操作,那么初始值又是如何进入每个线程成为独立拷贝的呢?首先,虽然ThreadLocal在定义时重写了initial

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

用户登录
用户注册