FileLock with timeout using Fcntl in Golang

file_lock_with_timeout.go

package main

import (
	"errors"
	"io"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"
)

var errBlocked = errors.New("acquiring lock is blocked by other process")

// lock locks file using FCNTL.
// FLOCK is not appropriate because FLOCK on darwin does not return EINTR in blocking mode.
func lock(file *os.File, nonblock bool) error {
	var cmd int
	if nonblock {
		cmd = syscall.F_SETLK
	} else {
		cmd = syscall.F_SETLKW
	}
	// write lock whole file
	flock := syscall.Flock_t{
		Start:  0,
		Len:    0,
		Type:   syscall.F_WRLCK,
		Whence: io.SeekStart,
	}

	if err := syscall.FcntlFlock(file.Fd(), cmd, &flock); err != nil {
		// FCNTL returns EINTR if interrupted by signal on blocking mode
		if !nonblock && err == syscall.EINTR {
			return errBlocked
		}
		// FCNTL returns EAGAIN or EACCESS if other process have locked the file on non-blocking
		if err == syscall.EAGAIN || err == syscall.EACCES {
			return errBlocked
		}
		return &os.PathError{Op: "fcntl", Path: file.Name(), Err: err}
	}
	return nil
}

func main() {
	sigch := make(chan os.Signal, 1)
	signal.Ignore()
	signal.Notify(sigch, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigch)

	file, err := os.Create("./.lock")
	if err != nil {
		log.Panic(err)
	}

	// init pthread
	tid := newPthread(syscall.SIGUSR1)

	// init error channel
	chErr := make(chan error, 1)

	// setup timer
	timer := time.NewTimer(3 * time.Second)

	go func() {
		// setup the thread signal settings
		if terr := tid.setup(); terr != nil {
			chErr <- terr
			return
		}
		defer func() {
			// reset signal settings
			if terr := tid.close(); terr != nil {
				// if failed to reset sigaction, go runtime will be broken.
				// terr occurs on C memory error which does not happen.
				panic(terr)
			}
		}()
		// lock file blocking
		chErr <- lock(file, false)
	}()

	for {
		select {
		case err = <-chErr:
			timer.Stop()
			if err == nil {
				log.Println("lock success")
			} else {
				log.Println("lock fail err", err)
			}
			// break loop
			return

		case <-timer.C:
			log.Println("timeout")

			// send signal to the thread locking file and unblock the lock with EINTR
			err := tid.signal()
			log.Println("signal")
			if err != nil {
				log.Panic("failed to kill thread", err)
			}
			// wait for lock result from chErr

		case sig := <-sigch:
			log.Println("signal", sig)
			// not call file.close() but it will blocks
			return
		}
	}
}

pthread.go

package main

/*
#include <stdio.h>
#include <stddef.h>
#include <errno.h>
#include <string.h>
#include <signal.h>
#include <pthread.h>

void sighandler(int sig){
	// empty handler
}

static pthread_t tid;
static struct sigaction oact;

static int setup_signal(int sig) {
	struct sigaction act;
	// set self thread id
	tid = pthread_self();

	// setup sigaction
	act.sa_handler = sighandler;
	act.sa_flags = 0;
	sigemptyset(&act.sa_mask);

	// set sigaction and cache old sigaction to oact
	if(sigaction(sig, &act, &oact) != 0){
		return errno;
	}
	return 0;
}

static int kill_thread(int sig) {
	// send signal to thread
	if (pthread_kill(tid, sig) == -1) {
		return errno;
	}
	return 0;
}

static int reset_signal(int sig) {
	// reset with old sigaction
	if(sigaction(sig, &oact, NULL) != 0){
		return errno;
	}
	return 0;
}
*/
import "C"
import (
	"fmt"
	"runtime"
	"sync"
	"syscall"
)

// return EFAULT when memory in C is invalid
// return EINVAL when signal is invalid
func setupSignal(sig syscall.Signal) error {
	ret := C.setup_signal(C.int(sig))
	if ret != 0 {
		return syscall.Errno(ret)
	}
	return nil
}

// return ESRCH when thread id is invalid
// return EINVAL when signal number is invalid
func killThread(sig syscall.Signal) error {
	ret := C.kill_thread(C.int(sig))
	if ret != 0 {
		return syscall.Errno(ret)
	}
	return nil
}

// return EFAULT when memory in C is invalid
// return EINVAL when signal is invalid
func resetSignal(sig syscall.Signal) error {
	ret := C.reset_signal(C.int(sig))
	if ret != 0 {
		return syscall.Errno(ret)
	}
	return nil
}

type pthread struct {
	mu     sync.Mutex
	cond   *sync.Cond
	sig    syscall.Signal
	init   bool
	closed bool
}

func newPthread(sig syscall.Signal) *pthread {
	p := &pthread{
		sig: sig,
	}
	p.cond = sync.NewCond(&p.mu)
	return p
}

// setup locks calling goroutine to this OS thread and overwrite sigaction with empty sa_flags (not including SA_RESTART)
// DO NOT call signal package functions until call (*pthread).close() or destroy sigaction with Golang runtime settings
func (p *pthread) setup() error {
	p.mu.Lock()
	defer func() {
		p.cond.Broadcast()
		p.mu.Unlock()
	}()

	// set initialized flag
	p.init = true

	// lock this goroutine to this os thread
	runtime.LockOSThread()

	// set thread id to global variable
	// overwrite sigaction of p.sig with empty handler to remove SA_RESTART
	err := setupSignal(p.sig)
	if err != nil {
		p.closed = true
		return fmt.Errorf("setup sigaction : %v", err)
	}
	return nil
}

func (p *pthread) close() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.closed {
		return nil
	}

	// unlock goroutine from os thread
	runtime.UnlockOSThread()

	// reset sigaction of p.sig with original value which is set by Golang runtime.
	if err := resetSignal(p.sig); err != nil {
		return fmt.Errorf("reset sigaction : %v", err)
	}
	p.closed = true
	return nil
}

func (p *pthread) signal() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if !p.init {
		// wait until setup finishes.
		p.cond.Wait()
	}
	if p.closed {
		return nil
	}

	// send p.sig signal to the thread.
	if err := killThread(p.sig); err != nil {
		return fmt.Errorf("send signal to thread : %v", err)
	}
	return nil
}