<commit_before>package main

import (
	"bufio"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sync"

	"github.com/flynn/flynn/Godeps/_workspace/src/github.com/technoweenie/grohl"
	"github.com/flynn/flynn/host/types"
)

type attachHandler struct {
	state   *State
	backend Backend
}

func (h *attachHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	var attachReq host.AttachReq
	if err := json.NewDecoder(req.Body).Decode(&attachReq); err != nil {
		http.Error(w, "invalid JSON", 400)
		return
	}
	conn, _, err := w.(http.Hijacker).Hijack()
	if err != nil {
		return
	}
	conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Type: application/vnd.flynn.attach-hijack\r\n\r\n"))
	h.attach(&attachReq, conn)
}

func (h *attachHandler) attach(req *host.AttachReq, conn io.ReadWriteCloser) {
	defer conn.Close()

	g := grohl.NewContext(grohl.Data{"fn": "attach", "job.id": req.JobID})
	g.Log(grohl.Data{"at": "start"})
	attachWait := make(chan struct{})
	job := h.state.AddAttacher(req.JobID, attachWait)
	if job == nil {
		defer h.state.RemoveAttacher(req.JobID, attachWait)
		if _, err := conn.Write([]byte{host.AttachWaiting}); err != nil {
			return
		}
		// TODO: add timeout
		g.Log(grohl.Data{"at": "wait"})
		<-attachWait
		job = h.state.GetJob(req.JobID)
	}
	w := bufio.NewWriter(conn)
	writeError := func(err string) {
		w.WriteByte(host.AttachError)
		binary.Write(w, binary.BigEndian, uint32(len(err)))
		w.WriteString(err)
		w.Flush()
	}
	if job.Status == host.StatusFailed {
		close(attachWait)
		writeError(*job.Error)
		return
	}

	writeMtx := &sync.Mutex{}
	writeMtx.Lock()

	attached := make(chan struct{})
	failed := make(chan struct{})
	opts := &AttachRequest{
		Job:      job,
		Logs:     req.Flags&host.AttachFlagLogs != 0,
		Stream:   req.Flags&host.AttachFlagStream != 0,
		Height:   req.Height,
		Width:    req.Width,
		Attached: attached,
	}
	var stdinW *io.PipeWriter
	if req.Flags&host.AttachFlagStdin != 0 {
		opts.Stdin, stdinW = io.Pipe()
	}
	if req.Flags&host.AttachFlagStdout != 0 {
		opts.Stdout = newFrameWriter(1, w, writeMtx)
	}
	if req.Flags&host.AttachFlagStderr != 0 {
		opts.Stderr = newFrameWriter(2, w, writeMtx)
	}

	go func() {
		defer func() {
			if stdinW != nil {
				stdinW.Close()
			}
		}()

		select {
		case <-attached:
			g.Log(grohl.Data{"at": "success"})
			conn.Write([]byte{host.AttachSuccess})
			writeMtx.Unlock()
			close(attached)
		case <-failed:
			g.Log(grohl.Data{"at": "failed"})
			return
		}
		close(attachWait)
		r := bufio.NewReader(conn)
		var buf [4]byte

		for {
			frameType, err := r.ReadByte()
			if err != nil {
				// TODO: signal close to attach and close all connections
				return
			}
			switch frameType {
			case host.AttachData:<commit_msg>host: Unlock write mutex when attach fails<commit_after>package main

import (
	"bufio"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sync"

	"github.com/flynn/flynn/Godeps/_workspace/src/github.com/technoweenie/grohl"
	"github.com/flynn/flynn/host/types"
)

type attachHandler struct {
	state   *State
	backend Backend
}

func (h *attachHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	var attachReq host.AttachReq
	if err := json.NewDecoder(req.Body).Decode(&attachReq); err != nil {
		http.Error(w, "invalid JSON", 400)
		return
	}
	conn, _, err := w.(http.Hijacker).Hijack()
	if err != nil {
		return
	}
	conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Type: application/vnd.flynn.attach-hijack\r\n\r\n"))
	h.attach(&attachReq, conn)
}

func (h *attachHandler) attach(req *host.AttachReq, conn io.ReadWriteCloser) {
	defer conn.Close()

	g := grohl.NewContext(grohl.Data{"fn": "attach", "job.id": req.JobID})
	g.Log(grohl.Data{"at": "start"})
	attachWait := make(chan struct{})
	job := h.state.AddAttacher(req.JobID, attachWait)
	if job == nil {
		defer h.state.RemoveAttacher(req.JobID, attachWait)
		if _, err := conn.Write([]byte{host.AttachWaiting}); err != nil {
			return
		}
		// TODO: add timeout
		g.Log(grohl.Data{"at": "wait"})
		<-attachWait
		job = h.state.GetJob(req.JobID)
	}
	w := bufio.NewWriter(conn)
	writeError := func(err string) {
		w.WriteByte(host.AttachError)
		binary.Write(w, binary.BigEndian, uint32(len(err)))
		w.WriteString(err)
		w.Flush()
	}
	if job.Status == host.StatusFailed {
		close(attachWait)
		writeError(*job.Error)
		return
	}

	writeMtx := &sync.Mutex{}
	writeMtx.Lock()

	attached := make(chan struct{})
	failed := make(chan struct{})
	opts := &AttachRequest{
		Job:      job,
		Logs:     req.Flags&host.AttachFlagLogs != 0,
		Stream:   req.Flags&host.AttachFlagStream != 0,
		Height:   req.Height,
		Width:    req.Width,
		Attached: attached,
	}
	var stdinW *io.PipeWriter
	if req.Flags&host.AttachFlagStdin != 0 {
		opts.Stdin, stdinW = io.Pipe()
	}
	if req.Flags&host.AttachFlagStdout != 0 {
		opts.Stdout = newFrameWriter(1, w, writeMtx)
	}
	if req.Flags&host.AttachFlagStderr != 0 {
		opts.Stderr = newFrameWriter(2, w, writeMtx)
	}

	go func() {
		defer func() {
			if stdinW != nil {
				stdinW.Close()
			}
		}()

		select {
		case <-attached:
			g.Log(grohl.Data{"at": "success"})
			conn.Write([]byte{host.AttachSuccess})
			writeMtx.Unlock()
			close(attached)
		case <-failed:
			g.Log(grohl.Data{"at": "failed"})
			writeMtx.Unlock()
			return
		}
		close(attachWait)
		r := bufio.NewReader(conn)
		var buf [4]byte

		for {
			frameType, err := r.ReadByte()
			if err != nil {
				// TODO: signal close to attach and close all connections
				return
			}
			switch frameType {
			case host.AttachData:
{'fraction_non_alphanumeric': 0.1614884393063584, 'fraction_numerical': 0.006502890173410405, 'mean_word_length': 3.652941176470588, 'pattern_counts': {'":': 12, '<': 9, '<?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': '3460514', 'n_tokens_mistral': 2360, 'n_tokens_neox': 2155, 'n_words': 420}