<commit_before>package containerd

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	goruntime "runtime"
	"strings"
	"syscall"

	eventsapi "github.com/containerd/containerd/api/services/events/v1"
	"github.com/containerd/containerd/api/services/tasks/v1"
	"github.com/containerd/containerd/api/types"
	"github.com/containerd/containerd/content"
	"github.com/containerd/containerd/errdefs"
	"github.com/containerd/containerd/linux/runcopts"
	"github.com/containerd/containerd/mount"
	"github.com/containerd/containerd/rootfs"
	"github.com/containerd/containerd/runtime"
	"github.com/containerd/containerd/typeurl"
	digest "github.com/opencontainers/go-digest"
	"github.com/opencontainers/image-spec/specs-go/v1"
	specs "github.com/opencontainers/runtime-spec/specs-go"
	"github.com/pkg/errors"
)

const UnknownExitStatus = 255

type Status struct {
	Status     ProcessStatus
	ExitStatus uint32
}

type ProcessStatus string

const (
	Running ProcessStatus = "running"
	Created ProcessStatus = "created"
	Stopped ProcessStatus = "stopped"
	Paused  ProcessStatus = "paused"
	Pausing ProcessStatus = "pausing"
)

type IOCloseInfo struct {
	Stdin bool
}

type IOCloserOpts func(*IOCloseInfo)

func WithStdinCloser(r *IOCloseInfo) {
	r.Stdin = true
}

type CheckpointTaskInfo struct {
	ParentCheckpoint digest.Digest
	Options          interface{}
}

type CheckpointTaskOpts func(*CheckpointTaskInfo) error

type TaskInfo struct {
	Checkpoint *types.Descriptor
	RootFS     []mount.Mount
	Options    interface{}
}

type Task interface {
	Pid() uint32
	Delete(context.Context) (uint32, error)
	Kill(context.Context, syscall.Signal) error
	Pause(context.Context) error
	Resume(context.Context) error
	Start(context.Context) error
	Status(context.Context) (Status, error)
	Wait(context.Context) (uint32, error)
	Exec(context.Context, string, *specs.Process, IOCreation) (Process, error)
	Pids(context.Context) ([]uint32, error)
	CloseIO(context.Context, ...IOCloserOpts) error
	Resize(ctx context.Context, w, h uint32) error
	IO() *IO
	Checkpoint(context.Context, ...CheckpointTaskOpts) (v1.Descriptor, error)
	Update(context.Context, ...UpdateTaskOpts) error
}

var _ = (Task)(&task{})

type task struct {
	client *Client

	io  *IO
	id  string
	pid uint32

	deferred *tasks.CreateTaskRequest
}

// Pid returns the pid or process id for the task
func (t *task) Pid() uint32 {
	return t.pid
}

func (t *task) Start(ctx context.Context) error {
	if t.deferred != nil {
		response, err := t.client.TaskService().Create(ctx, t.deferred)
		t.deferred = nil
		if err != nil {
			t.io.closer.Close()
			return err
		}
		t.pid = response.Pid
		return nil
	}
	_, err := t.client.TaskService().Start(ctx, &tasks.StartRequest{
		ContainerID: t.id,
	})
	if err != nil {
		t.io.closer.Close()
	}
	return err
}

func (t *task) Kill(ctx context.Context, s syscall.Signal) error {
	_, err := t.client.TaskService().Kill(ctx, &tasks.KillRequest{
		Signal:      uint32(s),
		ContainerID: t.id,
	})
	if err != nil {
		return errdefs.FromGRPC(err)
	}
	return nil
}

func (t *task) Pause(ctx context.Context) error {
	_, err := t.client.TaskService().Pause(ctx, &tasks.PauseTaskRequest{
		ContainerID: t.id,
	})
	return errdefs.FromGRPC(err)
}

func (t *task) Resume(ctx context.Context) error {
	_, err := t.client.TaskService().Resume(ctx, &tasks.ResumeTaskRequest{
		ContainerID: t.id,
	})
	return errdefs.FromGRPC(err)
}

func (t *task) Status(ctx context.Context) (Status, error) {
	r, err := t.client.TaskService().Get(ctx, &tasks.GetRequest{
		ContainerID: t.id,
	})
	if err != nil {
		return Status{}, errdefs.FromGRPC(err)
	}
	return Status{
		Status:     ProcessStatus(strings.ToLower(r.Process.Status.String())),
		ExitStatus: r.Process.ExitStatus,
	}, nil
}

func (t *task) Wait(ctx context.Context) (uint32, error) {
	eventstream, err := t.client.EventService().Subscribe(ctx, &eventsapi.SubscribeRequest{
		Filters: []string{"topic==" + runtime.TaskExitEventTopic},
	})
	if err != nil {
		return UnknownExitStatus, errdefs.FromGRPC(err)
	}
	// first check if the task has exited
	status, err := t.Status(ctx)
	if err != nil {
		return UnknownExitStatus, errdefs.FromGRPC(err)
	}
	if status.Status == Stopped {
		return status.ExitStatus, nil
	}
	for {
		evt, err := eventstream.Recv()
		if err != nil {<commit_msg>Gate task.Wait status check on checkpoints<commit_after>package containerd

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	goruntime "runtime"
	"strings"
	"sync"
	"syscall"

	eventsapi "github.com/containerd/containerd/api/services/events/v1"
	"github.com/containerd/containerd/api/services/tasks/v1"
	"github.com/containerd/containerd/api/types"
	"github.com/containerd/containerd/content"
	"github.com/containerd/containerd/errdefs"
	"github.com/containerd/containerd/linux/runcopts"
	"github.com/containerd/containerd/mount"
	"github.com/containerd/containerd/rootfs"
	"github.com/containerd/containerd/runtime"
	"github.com/containerd/containerd/typeurl"
	digest "github.com/opencontainers/go-digest"
	"github.com/opencontainers/image-spec/specs-go/v1"
	specs "github.com/opencontainers/runtime-spec/specs-go"
	"github.com/pkg/errors"
)

const UnknownExitStatus = 255

type Status struct {
	Status     ProcessStatus
	ExitStatus uint32
}

type ProcessStatus string

const (
	Running ProcessStatus = "running"
	Created ProcessStatus = "created"
	Stopped ProcessStatus = "stopped"
	Paused  ProcessStatus = "paused"
	Pausing ProcessStatus = "pausing"
)

type IOCloseInfo struct {
	Stdin bool
}

type IOCloserOpts func(*IOCloseInfo)

func WithStdinCloser(r *IOCloseInfo) {
	r.Stdin = true
}

type CheckpointTaskInfo struct {
	ParentCheckpoint digest.Digest
	Options          interface{}
}

type CheckpointTaskOpts func(*CheckpointTaskInfo) error

type TaskInfo struct {
	Checkpoint *types.Descriptor
	RootFS     []mount.Mount
	Options    interface{}
}

type Task interface {
	Pid() uint32
	Delete(context.Context) (uint32, error)
	Kill(context.Context, syscall.Signal) error
	Pause(context.Context) error
	Resume(context.Context) error
	Start(context.Context) error
	Status(context.Context) (Status, error)
	Wait(context.Context) (uint32, error)
	Exec(context.Context, string, *specs.Process, IOCreation) (Process, error)
	Pids(context.Context) ([]uint32, error)
	CloseIO(context.Context, ...IOCloserOpts) error
	Resize(ctx context.Context, w, h uint32) error
	IO() *IO
	Checkpoint(context.Context, ...CheckpointTaskOpts) (v1.Descriptor, error)
	Update(context.Context, ...UpdateTaskOpts) error
}

var _ = (Task)(&task{})

type task struct {
	client *Client

	io  *IO
	id  string
	pid uint32

	mu       sync.Mutex
	deferred *tasks.CreateTaskRequest
}

// Pid returns the pid or process id for the task
func (t *task) Pid() uint32 {
	return t.pid
}

func (t *task) Start(ctx context.Context) error {
	t.mu.Lock()
	deferred := t.deferred
	t.mu.Unlock()
	if deferred != nil {
		response, err := t.client.TaskService().Create(ctx, deferred)
		t.mu.Lock()
		t.deferred = nil
		t.mu.Unlock()
		if err != nil {
			t.io.closer.Close()
			return err
		}
		t.pid = response.Pid
		return nil
	}
	_, err := t.client.TaskService().Start(ctx, &tasks.StartRequest{
		ContainerID: t.id,
	})
	if err != nil {
		t.io.closer.Close()
	}
	return err
}

func (t *task) Kill(ctx context.Context, s syscall.Signal) error {
	_, err := t.client.TaskService().Kill(ctx, &tasks.KillRequest{
		Signal:      uint32(s),
		ContainerID: t.id,
	})
	if err != nil {
		return errdefs.FromGRPC(err)
	}
	return nil
}

func (t *task) Pause(ctx context.Context) error {
	_, err := t.client.TaskService().Pause(ctx, &tasks.PauseTaskRequest{
		ContainerID: t.id,
	})
	return errdefs.FromGRPC(err)
}

func (t *task) Resume(ctx context.Context) error {
	_, err := t.client.TaskService().Resume(ctx, &tasks.ResumeTaskRequest{
		ContainerID: t.id,
	})
	return errdefs.FromGRPC(err)
}

func (t *task) Status(ctx context.Context) (Status, error) {
	r, err := t.client.TaskService().Get(ctx, &tasks.GetRequest{
		ContainerID: t.id,
	})
	if err != nil {
		return Status{}, errdefs.FromGRPC(err)
	}
	return Status{
		Status:     ProcessStatus(strings.ToLower(r.Process.Status.String())),
		ExitStatus: r.Process.ExitStatus,
	}, nil
}

func (t *task) Wait(ctx context.Context) (uint32, error) {
	eventstream, err := t.client.EventService().Subscribe(ctx, &eventsapi.SubscribeRequest{
		Filters: []string{"topic==" + runtime.TaskExitEventTopic},
	})
	if err != nil {
		return UnknownExitStatus, errdefs.FromGRPC(err)
	}
	t.mu.Lock()
	checkpoint := t.deferred != nil
	t.mu.Unlock()
	if !checkpoint {
		// first check if the task has exited
		status, err := t.Status(ctx)
		if err != nil {
			return UnknownExitStatus, errdefs.FromGRPC(err)
		}
		if status.Status == Stopped {
			return status.ExitStatus, nil
		}
	}
	for {
		evt, err := eventstream.Recv()
		if err != nil {
{'fraction_non_alphanumeric': 0.1350158442734269, 'fraction_numerical': 0.0061113626075147125, 'mean_word_length': 4.333132166566084, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3236341', 'n_tokens_mistral': 3307, 'n_tokens_neox': 3185, 'n_words': 685}