add k-core algorithm

computer-algorithm/src/main/java/com/baidu/hugegraph/computer/algorithm/community/kcore/KCore.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.computer.algorithm.community.kcore;
+
+import java.util.Iterator;
+import com.baidu.hugegraph.computer.core.config.Config;
+import com.baidu.hugegraph.computer.core.graph.id.Id;
+import com.baidu.hugegraph.computer.core.graph.vertex.Vertex;
+import com.baidu.hugegraph.computer.core.worker.Computation;
+import com.baidu.hugegraph.computer.core.worker.ComputationContext;
+
+public class KCore implements Computation<Id> {
+
+    public static final String ALGORITHM_NAME = "kcore";
+    public static final String OPTION_K = "kcore.k";
+    public static final int K_DEFAULT_VALUE = 3;
+
+    private final KCoreValue initValue = new KCoreValue();
+
+    private int k;
+
+    @Override
+    public String name() {
+        return ALGORITHM_NAME;
+    }
+
+    @Override
+    public String category() {
+        return "community";
+    }
+
+    @Override
+    public void init(Config config) {
+        this.k = config.getInt(OPTION_K, K_DEFAULT_VALUE);
+    }
+
+    @Override
+    public void compute0(ComputationContext context, Vertex vertex) {
+        KCoreValue value = this.initValue;
+        value.core(vertex.numEdges());
+        vertex.value(value);
+
+        if (value.core() < this.k) {
+            value.core(0);
+            context.sendMessageToAllEdges(vertex, vertex.id());
+        }
+        vertex.inactivate();
+    }
+
+    @Override
+    public void compute(ComputationContext context, Vertex vertex,
+                        Iterator<Id> messages) {
+        KCoreValue value = vertex.value();
+        if (!value.active()) {
+            vertex.inactivate();
+            return;
+        }
+
+        int deleted = 0;
+        while (messages.hasNext()) {
+            Id neighborId = messages.next();
+            value.addDeletedNeighbor(neighborId);
+            deleted++;
+        }
+
+        if (value.decreaseCore(deleted) < this.k) {
+            value.core(0);
+            context.sendMessageToAllEdgesIf(
+                    vertex, vertex.id(),
+                    (source, target) -> {
+                        return !value.isNeighborDeleted(target);
+                    });
+        }
+        vertex.inactivate();
+    }
+}

computer-algorithm/src/main/java/com/baidu/hugegraph/computer/algorithm/community/kcore/KCoreOutput.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.computer.algorithm.community.kcore;
+
+import com.baidu.hugegraph.computer.core.output.hg.HugeOutput;
+import com.baidu.hugegraph.structure.constant.WriteType;
+import com.baidu.hugegraph.structure.graph.Vertex;
+
+public class KCoreOutput extends HugeOutput {
+
+    @Override
+    public String name() {
+        return KCore.ALGORITHM_NAME;
+    }
+
+    @Override
+    public void prepareSchema() {
+        this.client().schema().propertyKey(this.name())
+            .asInt()
+            .writeType(WriteType.OLAP_COMMON)
+            .ifNotExist()
+            .create();
+    }
+
+    @Override
+    public Vertex constructHugeVertex(
+           com.baidu.hugegraph.computer.core.graph.vertex.Vertex vertex) {
+        com.baidu.hugegraph.structure.graph.Vertex hugeVertex =
+                new com.baidu.hugegraph.structure.graph.Vertex(null);
+        hugeVertex.id(vertex.id().asObject());
+        KCoreValue value = vertex.value();
+        hugeVertex.property(this.name(), value.core());
+        return hugeVertex;
+    }
+}

computer-algorithm/src/main/java/com/baidu/hugegraph/computer/algorithm/community/kcore/KCoreParams.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.computer.algorithm.community.kcore;
+
+import java.util.Map;
+
+import com.baidu.hugegraph.computer.algorithm.AlgorithmParams;
+import com.baidu.hugegraph.computer.core.config.ComputerOptions;
+import com.baidu.hugegraph.computer.core.graph.id.BytesId;
+
+public class KCoreParams implements AlgorithmParams {
+
+    @Override
+    public void setAlgorithmParameters(Map<String, String> params) {
+        this.setIfAbsent(params, ComputerOptions.WORKER_COMPUTATION_CLASS,
+                         KCore.class.getName());
+        this.setIfAbsent(params, ComputerOptions.ALGORITHM_RESULT_CLASS,
+                         KCoreValue.class.getName());
+        this.setIfAbsent(params, ComputerOptions.ALGORITHM_MESSAGE_CLASS,
+                         BytesId.class.getName());
+        this.setIfAbsent(params, ComputerOptions.OUTPUT_CLASS,
+                         KCoreOutput.class.getName());
+
+        params.put(ComputerOptions.VERTEX_WITH_EDGES_BOTHDIRECTION.name(),
+                   Boolean.TRUE.toString());
+        params.put(ComputerOptions.BSP_MAX_SUPER_STEP.name(),
+                   String.valueOf(Integer.MAX_VALUE));
+    }
+}

computer-algorithm/src/main/java/com/baidu/hugegraph/computer/algorithm/community/kcore/KCoreValue.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2017 HugeGraph Authors
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.baidu.hugegraph.computer.algorithm.community.kcore;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+import com.baidu.hugegraph.computer.core.graph.id.BytesId;
+import com.baidu.hugegraph.computer.core.graph.id.Id;
+import com.baidu.hugegraph.computer.core.graph.value.Value;
+import com.baidu.hugegraph.computer.core.graph.value.ValueType;
+import com.baidu.hugegraph.computer.core.io.RandomAccessInput;
+import com.baidu.hugegraph.computer.core.io.RandomAccessOutput;
+
+public class KCoreValue implements Value<KCoreValue> {
+
+    private int core;
+    private Set<Id> deletedNeighbors;
+
+    public KCoreValue() {
+        this.core = 0;
+        this.deletedNeighbors = new HashSet<>();
+    }
+
+    public void core(int core) {
+        assert core >= 0;
+        this.core = core;
+    }
+
+    public int core() {
+        return this.core;
+    }
+
+    public int decreaseCore(int decrease) {
+        assert decrease <= this.core;
+        this.core -= decrease;
+        return this.core;
+    }
+
+    public boolean active() {
+        return this.core > 0;
+    }
+
+    public boolean isNeighborDeleted(Id neighborId) {
+        return this.deletedNeighbors.contains(neighborId);
+    }
+
+    public void addDeletedNeighbor(Id neighborId) {
+        this.deletedNeighbors.add(neighborId);
+    }
+
+    @Override
+    public ValueType valueType() {
+        return ValueType.UNKNOWN;
+    }
+
+    @Override
+    public void assign(Value<KCoreValue> other) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Value<KCoreValue> copy() {
+        KCoreValue kcoreValue = new KCoreValue();
+        kcoreValue.core = this.core;
+        return kcoreValue;
+    }
+
+    @Override
+    public void read(RandomAccessInput in) throws IOException {
+        this.core = in.readInt();
+        int size = in.readInt();
+        this.deletedNeighbors = new HashSet<>((int)((float) size / 0.75 + 1));
+        for (int i = 0; i < size; i++) {
+            Id id = new BytesId();
+            id.read(in);
+            deletedNeighbors.add(id);
+        }
+    }
+
+    @Override
+    public void write(RandomAccessOutput out) throws IOException {
+        out.writeInt(this.core);
+        out.writeInt(this.deletedNeighbors.size());
+        for (Id id : this.deletedNeighbors) {
+            id.write(out);
+        }
+    }
+
+    @Override
+    public int compareTo(KCoreValue other) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public String toString() {
+        return new ToStringBuilder(this)
+                   .append("core", this.core)
+                   .toString();
+    }
+
+    @Override
+    public String string() {
+        return String.valueOf(this.value());
+    }
+
+    @Override
+    public Object value() {
+        return this.core;
+    }
+}

computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/FileGraphPartition.java

@@ -161,6 +161,7 @@ protected PartitionStat compute0(ComputationContext context,
             Vertex vertex = this.vertexInput.next();
             Edges edges = this.edgesInput.edges(this.vertexInput.idPointer());
             vertex.edges(edges);
+            vertex.reactivate();
             computation.compute0(context, vertex);
             if (vertex.active()) {
                 activeVertexCount++;

computer-core/src/main/java/com/baidu/hugegraph/computer/core/compute/input/MessageInput.java

@@ -43,6 +43,7 @@ public class MessageInput<T extends Value<?>> {
     private final Config config;
     private final PeekableIterator<KvEntry> messages;
     private T value;
+    private boolean inCompute;
 
     public MessageInput(ComputerContext context,
                         PeekableIterator<KvEntry> messages,
@@ -53,6 +54,7 @@ public MessageInput(ComputerContext context,
             this.messages = messages;
         }
         this.config = context.config();
+        this.inCompute = inCompute;
 
         if (!inCompute) {
             this.value = (T)(new IdList());
@@ -99,7 +101,6 @@ public Iterator<T> iterator(ReusablePointer vidPointer) {
         while (this.messages.hasNext()) {
             KvEntry entry = this.messages.peek();
             Pointer key = entry.key();
-            Pointer value = entry.value();
             int status = vidPointer.compareTo(key);
             if (status < 0) {
                 return Collections.emptyIterator();
@@ -143,6 +144,15 @@ public boolean hasNext() {
                     try {
                         BytesInput in = IOFactory.createBytesInput(
                                         entry.value().bytes());
+                        /*
+                         * TODO: solve the problem of combine shared objects
+                         *  in another way
+                         */
+                        if (MessageInput.this.inCompute) {
+                            MessageInput.this.value =
+                            MessageInput.this.config.createObject(
+                            ComputerOptions.ALGORITHM_MESSAGE_CLASS);
+                        }
                         MessageInput.this.value.read(in);
                     } catch (IOException e) {
                         throw new ComputerException("Can't read value", e);

computer-test/src/main/java/com/baidu/hugegraph/computer/algorithm/AlgorithmTestSuite.java

@@ -27,6 +27,7 @@
 import com.baidu.hugegraph.computer.algorithm.centrality.closeness.ClosenessCentralityTest;
 import com.baidu.hugegraph.computer.algorithm.centrality.degree.DegreeCentralityTest;
 import com.baidu.hugegraph.computer.algorithm.centrality.pagerank.PageRankTest;
+import com.baidu.hugegraph.computer.algorithm.community.kcore.KCoreTest;
 import com.baidu.hugegraph.computer.algorithm.community.lpa.LpaTest;
 import com.baidu.hugegraph.computer.algorithm.community.trianglecount.TriangleCountTest;
 import com.baidu.hugegraph.computer.algorithm.community.wcc.WccTest;
@@ -52,7 +53,8 @@
     BetweennessCentralityTest.class,
     LinksTest.class,
     SsspTest.class,
-    SubGraphMatchTest.class
+    SubGraphMatchTest.class,
+    KCoreTest.class
 })
 public class AlgorithmTestSuite {