go - 为什么下面的 golang 程序会抛出运行时内存不足错误？

Question

这个程序应该读取一个由整数对组成的文件(每行一对)并删除重复的对。虽然它适用于小文件，但它会在大文件(比如 1.5 GB 的文件)上引发运行时错误。最初，我认为是 map 数据结构导致的，但即使将其注释掉，它仍然会耗尽内存。任何想法为什么会这样？如何纠正？这是一个内存不足的数据文件:http://snap.stanford.edu/data/com-Orkut.html

package main
import (
    "fmt"
    "bufio"
    "os"
    "strings"
    "strconv"
)

func main() {
    file, err := os.Open(os.Args[1])
    if err != nil {
        panic(err.Error())
    }
    defer file.Close()
    type Edge struct {
        u, v int
    }
    //seen := make(map[Edge]bool)
    edges := []Edge{}
    scanner := bufio.NewScanner(file)

    for i, _ := strconv.Atoi(os.Args[2]); i > 0; i-- {
        scanner.Scan()
    }

    for scanner.Scan() {
        str := scanner.Text()
        edge := strings.Split(str, ",")
        u, _ := strconv.Atoi(edge[0])
        v, _ := strconv.Atoi(edge[1])
        var key Edge
        if u < v {
            key = Edge{u,v}
        } else {
            key = Edge{v,u}
        }
        //if seen[key] {
        //  continue
        //}
        //seen[key] = true
        edges = append(edges, key)
    }
    for _, e := range edges {
        s := strconv.Itoa(e.u) + "," + strconv.Itoa(e.v)
        fmt.Println(s)
    }
}

下面给出了示例输入。该程序可以按如下方式运行(最后一个输入表示要跳过多少行)。去运行 undup.go a.txt 1

# 3072441,117185083
1,2
1,3
1,4
1,5
1,6
1,7
1,8

Answer 1

我查看了这个文件:com-orkut.ungraph.txt，它包含 117,185,082 行。您的数据结构方式，每行至少 16 个字节。 (Edge 是两个 64 位整数)仅此一项就是 1.7GB。我过去遇到过这个问题，这可能是一个棘手的问题。您是要针对特定​​用例(有问题的文件)还是一般情况来解决此问题？

在特定情况下，您可以利用一些关于数据的信息:(1) 键已排序，并且 (2) 它看起来将每个连接存储两次，(3) 数字似乎并不大。这里有几个想法:

1. 如果您使用较小的 key 类型，您将使用较少的内存。试试 uint32。

2. 您可以通过简单地查看第二列是否大于第一列来将 key 流式传输(不使用映射)到另一个文件:

   if u < v {
       // write the key to another file
   } else {
       // skip it because v will eventually show v -> u
   }
   

对于一般情况，您可以使用几种策略:

1. 如果结果列表的顺序无关紧要:使用磁盘上的哈希表来存储 map 。有很多这样的东西:leveldb、sqlite、tokyo tyrant……一个非常好的 go 是 bolt .

   在您的 for 循环中，您只需检查一个桶是否包含给定的键。 (您可以使用编码/二进制将整数转换为 byte slice )如果是这样，请跳过它并继续。您需要将第二个 for 循环处理步骤移到第一个 for 循环中，这样您就不必存储所有 key 。

2. 如果结果列表的顺序很重要(并且您不能保证输入的顺序):您也可以使用磁盘上的哈希表，但需要对其进行排序。 bolt 已排序，因此可以使用。把所有的key都加进去，然后在第二个循环遍历。

这是一个例子:(这个程序需要一段时间才能运行 1 亿条记录)

package main

import (
    "bufio"
    "encoding/binary"
    "fmt"
    "github.com/boltdb/bolt"
    "os"
    "strconv"
    "strings"
)

type Edge struct {
    u, v int
}

func FromKey(bs []byte) Edge {
    return Edge{int(binary.BigEndian.Uint64(bs[:8])), int(binary.BigEndian.Uint64(bs[8:]))}
}

func (e Edge) Key() [16]byte {
    var k [16]byte
    binary.BigEndian.PutUint64(k[:8], uint64(e.u))
    binary.BigEndian.PutUint64(k[8:], uint64(e.v))
    return k
}

func main() {
    file, err := os.Open(os.Args[1])
    if err != nil {
        panic(err.Error())
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)

    for i, _ := strconv.Atoi(os.Args[2]); i > 0; i-- {
        scanner.Scan()
    }

    db, _ := bolt.Open("ex.db", 0777, nil)
    defer db.Close()

    bucketName := []byte("edges")
    db.Update(func(tx *bolt.Tx) error {
        tx.CreateBucketIfNotExists(bucketName)
        return nil
    })

    batchSize := 10000
    total := 0
    batch := make([]Edge, 0, batchSize)
    writeBatch := func() {
        total += len(batch)
        fmt.Println("write batch. total:", total)
        db.Update(func(tx *bolt.Tx) error {
            bucket := tx.Bucket(bucketName)
            for _, edge := range batch {
                key := edge.Key()
                bucket.Put(key[:], nil)
            }
            return nil
        })
    }

    for scanner.Scan() {
        str := scanner.Text()
        edge := strings.Split(str, "\t")
        u, _ := strconv.Atoi(edge[0])
        v, _ := strconv.Atoi(edge[1])
        var key Edge
        if u < v {
            key = Edge{u, v}
        } else {
            key = Edge{v, u}
        }
        batch = append(batch, key)
        if len(batch) == batchSize {
            writeBatch()
            // reset the batch length to 0
            batch = batch[:0]
        }
    }
    // write anything leftover
    writeBatch()

    db.View(func(tx *bolt.Tx) error {
        tx.Bucket(bucketName).ForEach(func(k, v []byte) error {
            edge := FromKey(k)
            fmt.Println(edge)
            return nil
        })
        return nil
    })
}