画布 3
content service
services/content/service.go
  func (s *Service) Write(session api.Content_WriteServer) (err error)
// 以pull docker.io/library/busybox:latest为例
    279 defer func(msg *api.WriteContentResponse) {
280 // pump through the last message if no error was encountered
281 if err != nil {
......
293 }
295 err = session.Send(msg)
296 }(&msg)
    298 // handle the very first request!
299 req, err = session.Recv()
// req = contentapi.WriteContentRequest{
Action: contentapi.WriteActionStat,
Ref: ref, // manifest-<digest>
Total: size, // 527
Expected: expected, // digest
}
    ref = req.Ref
total = req.Total
expected = req.Expected
    327 wr, err := s.store.Writer(ctx, ref, total, expected)
331 defer wr.Close()
      metadata/content.go
286 func (cs *contentStore) Writer(ctx context.Context, ref string, size int64, expected digest.Digest) (content.Writer, error) {}
// metadata store是contentPlugin的一个封装，content plugin是 content/local/store的一个封装
        287 ns, err := namespaces.NamespaceRequired(ctx)
        if err := update(ctx, cs.db, func(tx *bolt.Tx) error {
......
}; err != nil {}
          294 if expected != "" {
295 cbkt := getBlobBucket(tx, ns, expected)
299 }
            metadata/buckets.go:
158 func getBlobBucket(tx *bolt.Tx, namespace string, dgst digest.Digest) *bolt.Bucket {
159 return getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectBlob, []byte(dgst.String()))
160 }
              55 func getBucket(tx *bolt.Tx, keys ...[]byte) *bolt.Bucket {
56 bkt := tx.Bucket(keys[0])
57
58 for _, key := range keys[1:] {
59 if bkt == nil {
60 break
61 }
62 bkt = bkt.Bucket(key)
63 }
64
65 return bkt
66 }
          bkt, err := createIngestBucket(tx, ns)
            bkt, err := createBucketIfNotExists(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectIngest)
          308 brefb = bkt.Get([]byte(ref))
311 if brefb == nil {
312 sid, err := bkt.NextSequence()
317 bref = createKey(sid, ns, ref) // "<ns>/<sid>/manifest-<dgst>"
318 if err := bkt.Put([]byte(ref), []byte(bref)); err != nil {
// key:value = manifest-<dgst>:<ns>/<sid>/manifest-<dgst>
319 return err
320 }
321 } else {
322 bref = string(brefb)
323 }
          w, err = cs.Store.Writer(ctx, bref, size, "") // bref: "<ns>/<sid>/manifest-<dgst>"; size:527
            content/local/store.go:
273 func (s *store) Writer(ctx context.Context, ref string, total int64, expected digest.Digest) (content.Writer, error) {...}
              // 此处为空，跳过
276 if expected != "" {
277 p := s.blobPath(expected)
278 if _, err := os.Stat(p); ...
281 }
              path, refp, data := s.ingestPaths(ref)
                400 func (s *store) ingestPaths(ref string) (string, string, string) {
401 var (
402 fp = s.ingestRoot(ref) // "/var/lib/containerd/io.containerd.content.v1.content/ingest/<digest>"
403 rp = filepath.Join(fp, "ref")
404 dp = filepath.Join(fp, "data")
405 )
406
407 return fp, rp, dp
408 }
                  389 func (s *store) ingestRoot(ref string) string {
390 dgst := digest.FromString(ref) // 对"<ns>/<id>/manifest-xxx" 做sha256sum的hash值
391 return filepath.Join(s.root, "ingest", dgst.Hex())
392 }
              // 如果目录已经存在，则
status, err := s.status(path)
fp, err := os.Open(data)
offset, err = io.CopyBuffer(digester.Hash(), fp, p)
updatedAt = status.UpdatedAt
startedAt = status.StartedAt
total = status.Total
              // 如果目录不存在，创建并：
ioutil.WriteFile(refp, []byte(ref), 0666) // 把 ref写进ingest/<digest>/ref文件
ioutil.WriteFile(filepath.Join(path, "total"), []byte(fmt.Sprint(total)), 0666) // 把total数值写进去
startedAt = time.Now()
updatedAt = startedAt
              fp, err := os.OpenFile(data, os.O_WRONLY|os.O_CREATE, 0666)
// fp是 ingest/<digest>/data的写句柄
              357 return &writer{
358 s: s, // 指向自己的指针
359 fp: fp, // data文件句柄
360 ref: ref, // ref字符串
361 path: path, // ingest/<digest> 目录
362 offset: offset, // 新创建的话就是为0,旧的话就是已存在的长度
363 total: total, // 总长度
364 digester: digester, // SHA256 digester , 保存着data文件的hash值
365 startedAt: startedAt, // 文件创建时间
366 updatedAt: updatedAt, // 更新时间
367 }, nil
// note: content/local.writer实现了content.Writer接口
        metadata/content.go:
340 return &namespacedWriter{
341 Writer: w, // local file store writer. -- content/local.writer
342 ref: ref, // manifest-dgst
343 namespace: ns, // <default ns>
344 db: cs.db, // metadata boltdb
345 }, nil
// namespacedWriter也实现了content.Writer接口，是local.writer的二层封装
    333 for {
// 处理。处理过程见右侧
session.Send(&msg)
req, err = session.Recv() // 接收新请求，继续循环
372 }
      334 msg.Action = req.Action
335 ws, err := wr.Status()
340 msg.Offset = ws.Offset // always set the offset.
      359 if req.Expected != "" {
365 if _, err := s.store.Info(session.Context(), req.Expected); err == nil {
366 if err := s.store.Abort(session.Context(), ref); err != nil {
367 log.G(ctx).WithError(err).Error("failed to abort write")
368 }
370 return grpc.Errorf(codes.AlreadyExists, "blob with expected digest %v exists", req.Expected)
371 }
372 }
      383 switch req.Action {
384 case api.WriteActionStat:
385 msg.Digest = wr.Digest()
386 msg.StartedAt = ws.StartedAt
387 msg.UpdatedAt = ws.UpdatedAt
388 msg.Total = total
        // 直接透传到content/store/writer.go
41 func (w *writer) Digest() digest.Digest {
42 return w.digester.Digest()
43 }
      389 case api.WriteActionWrite, api.WriteActionCommit:
// 第一步offset设置成append或者从头0开始
404 // issue the write if we actually have data.
405 if len(req.Data) > 0 {
408 n, err := wr.Write(req.Data)
413 if n != len(req.Data) {
416 return grpc.Errorf(codes.DataLoss, "wrote %v of %v bytes", n, len(req.Data))
417 }
419 msg.Offset += int64(n)
420 }
422 if req.Action == api.WriteActionCommit {
423 var opts []content.Opt
424 if req.Labels != nil {
425 opts = append(opts, content.WithLabels(req.Labels))
426 }
427 if err := wr.Commit(total, expected, opts...); err != nil {
428 return err
429 }
430 }
432 msg.Digest = wr.Digest()
        // Write()同样透传给content/local/writer.go
49 func (w *writer) Write(p []byte) (n int, err error) {
50 n, err = w.fp.Write(p)
51 w.digester.Hash().Write(p[:n])
52 w.offset += int64(len(p))
53 w.updatedAt = time.Now()
54 return n, err
55 }