text:
package cmd

import (
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net/url"
	"os"
	pathutil "path"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/dustin/go-humanize"
	jsoniter "github.com/json-iterator/go"
	"github.com/klauspost/readahead"
	"github.com/minio/minio/internal/bucket/lifecycle"
	"github.com/minio/minio/internal/color"
	"github.com/minio/minio/internal/config"
	"github.com/minio/minio/internal/disk"
	xioutil "github.com/minio/minio/internal/ioutil"
	"github.com/minio/minio/internal/logger"
	"github.com/minio/pkg/console"
	"github.com/minio/pkg/env"
)

const (
	nullVersionID        = "null"
	blockSizeSmall       = 128 * humanize.KiByte // Default r/w block size for smaller objects.
	blockSizeLarge       = 2 * humanize.MiByte   // Default r/w block size for larger objects.
	blockSizeReallyLarge = 4 * humanize.MiByte   // Default write block size for objects per shard >= 64MiB

	// On regular files bigger than this;
	readAheadSize = 16 << 20
	// Read this many buffers ahead.
	readAheadBuffers = 4
	// Size of each buffer.
	readAheadBufSize = 1 << 20

	// Really large streams threshold per shard.
	reallyLargeFileThreshold = 64 * humanize.MiByte // Optimized for HDDs

	// Small file threshold below which data accompanies metadata from storage layer.
	smallFileThreshold = 128 * humanize.KiByte // Optimized for NVMe/SSDs
	// For hardrives it is possible to set this to a lower value to avoid any
	// spike in latency. But currently we are simply keeping it optimal for SSDs.

	// XL metadata file carries per object metadata.
	xlStorageFormatFile = "xl.meta"
)

var alignedBuf []byte

func init() {
	alignedBuf = disk.AlignedBlock(4096)
	_, _ = rand.Read(alignedBuf)
}

// isValidVolname verifies a volname name in accordance with object
// layer requirements.
func isValidVolname(volname string) bool {
	if len(volname) < 3 {
		return false
	}

	if runtime.GOOS == "windows" {
		// Volname shouldn't have reserved characters in Windows.
		return !strings.ContainsAny(volname, `\:*?\"<>|`)
	}

	return true
}

var (
	xlPoolReallyLarge = sync.Pool{
		New: func() interface{} {
			b := disk.AlignedBlock(blockSizeReallyLarge)
			return &b
		},
	}
	xlPoolLarge = sync.Pool{
		New: func() interface{} {
			b := disk.AlignedBlock(blockSizeLarge)
			return &b
		},
	}
	xlPoolSmall = sync.Pool{
		New: func() interface{} {
			b := disk.AlignedBlock(blockSizeSmall)
			return &b
		},
	}
)

// xlStorage - implements StorageAPI interface.
type xlStorage struct {
	diskPath string
	endpoint Endpoint

	globalSync bool

	rootDisk bool

	diskID string

	// Indexes, will be -1 until assigned a set.
	poolIndex, setIndex, diskIndex int

	formatFileInfo  os.FileInfo
	formatLegacy    bool
	formatLastCheck time.Time

	diskInfoCache timedValue

	ctx context.Context
	sync.RWMutex

	// mutex to prevent concurrent read operations overloading walks.
	walkMu sync.Mutex
}

// checkPathLength - returns error if given path name length more than 255
func checkPathLength(pathName string) error {
	// Apple OS X path length is limited to 1016
	if runtime.GOOS == "darwin" && len(pathName) > 1016 {
		return errFileNameTooLong
	}

	// Disallow more than 1024 characters on windows, there
	// are no known name_max limits on Windows.
	if runtime.GOOS == "windows" && len(pathName) > 1024 {
		return errFileNameTooLong
	}

	// On Unix we reject paths if they are just '.', '..' or '/'
	if pathName == "." || pathName == ".." || pathName == slashSeparator {
		return errFileAccessDenied
	}

	// Check each path segment length is > 255 on all Unix
	// platforms, look for this value as NAME_MAX in
	// /usr/include/linux/limits.h
	var count int64
	for _, p := range pathName {
		switch p {
		case '/':
			count = 0 // Reset
		case '\\':
			if runtime.GOOS == globalWindowsOSName {
				count = 0
			}
		default:
			count++
			if count > 255 {
				return errFileNameTooLong
			}
		}
	} // Success.
	return nil
}

func getValidPath(path string) (string, error) {
	if path == "" {
		return path, errInvalidArgument
	}

	var err error
	// Disallow relative paths, figure out absolute paths.
	path, err = filepath.Abs(path)
	if err != nil {
		return path, err
	}

	fi, err := Lstat(path)
	if err != nil && !osIsNotExist(err) {
		return path, err
	}
	if osIsNotExist(err) {
		// Disk not found create it.
		if err = mkdirAll(path, 0777); err != nil {
			return path, err
		}
	}
	if fi != nil && !fi.IsDir() {
		return path, errDiskNotDir
	}

	return path, nil
}

// isDirEmpty - returns whether given directory is empty or not.
func isDirEmpty(dirname string) bool {
	entries, err := readDirN(dirname, 1)
	if err != nil {
		if err != errFileNotFound {
			logger.LogIf(GlobalContext, err)
		}
		return false
	}
	return len(entries) == 0
}

// Initialize a new storage disk.
func newLocalXLStorage(path string) (*xlStorage, error) {
	u := url.URL{Path: path}
	return newXLStorage(Endpoint{
		URL:     &u,
		IsLocal: true,
	})
}

// Sanitize - sanitizes the `format.json`, cleanup tmp.
// all other future cleanups should be added here.
func (s *xlStorage) Sanitize() error {
	if err := formatErasureMigrate(s.diskPath); err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}

	return formatErasureCleanupTmp(s.diskPath)
}

// Initialize a new storage disk.
func newXLStorage(ep Endpoint) (*xlStorage, error) {
	path := ep.Path
	var err error
	if path, err = getValidPath(path); err != nil {
		return nil, err
	}

	var rootDisk bool
	if env.Get("MINIO_CI_CD", "") != "" {
		rootDisk = true
	} else {
		rootDisk, err = disk.IsRootDisk(path, SlashSeparator)
		if err != nil {
			return nil, err
		}
		if !rootDisk {
			// If for some reason we couldn't detect the
			// root disk use - MINIO_ROOTDISK_THRESHOLD_SIZE
			// to figure out if the disk is root disk or not.
			if rootDiskSize := env.Get(config.EnvRootDiskThresholdSize, ""); rootDiskSize != "" {
				info, err := disk.GetInfo(path)
				if err != nil {
					return nil, err
				}
				size, err := humanize.ParseBytes(rootDiskSize)
				if err != nil {
					return nil, err
				}
				// size of the disk is less than the threshold or
				// equal to the size of the disk at path, treat
				// such disks as rootDisks and reject them.
				rootDisk = info.Total <= size
			}
		}
	}

	p := &xlStorage{
		diskPath:   path,
		endpoint:   ep,
		globalSync: env.Get(config.EnvFSOSync, config.EnableOff) == config.EnableOn,
		ctx:        GlobalContext,
		rootDisk:   rootDisk,
		poolIndex:  -1,
		setIndex:   -1,
		diskIndex:  -1,
	}

	// Create all necessary bucket folders if possible.
	if err = p.MakeVolBulk(context.TODO(), minioMetaBucket, minioMetaTmpBucket, minioMetaMultipartBucket, dataUsageBucket, minioMetaSpeedTestBucket); err != nil {
		return nil, err
	}

	// Check if backend is writable and supports O_DIRECT
	var rnd [8]byte
	_, _ = rand.Read(rnd[:])
	tmpFile := ".writable-check-" + hex.EncodeToString(rnd[:]) + ".tmp"
	filePath := pathJoin(p.diskPath, minioMetaTmpBucket, tmpFile)
	w, err := OpenFileDirectIO(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0666)
	if err != nil {
		return p, err
	}
	if _, err = w.Write(alignedBuf[:]); err != nil {
		w.Close()
		return p, err
	}
	w.Close()
	Remove(filePath)

	// Success.
	return p, nil
}

// getDiskInfo returns given disk information.
func getDiskInfo(diskPath string) (di disk.Info, err error) {
	if err = checkPathLength(diskPath); err == nil {
		di, err = disk.GetInfo(diskPath)
	}

	switch {
	case osIsNotExist(err):
		err = errDiskNotFound
	case isSysErrTooLong(err):
		err = errFileNameTooLong
	case isSysErrIO(err):
		err = errFaultyDisk
	}

	return di, err
}

// Implements stringer compatible interface.
func (s *xlStorage) String() string {
	return s.diskPath
}

func (s *xlStorage) Hostname() string {
	return s.endpoint.Host
}

func (s *xlStorage) Endpoint() Endpoint {
	return s.endpoint
}

func (*xlStorage) Close() error {
	return nil
}

func (s *xlStorage) IsOnline() bool {
	return true
}

func (s *xlStorage) LastConn() time.Time {
	return time.Time{}
}

func (s *xlStorage) IsLocal() bool {
	return true
}

// Retrieve location indexes.
func (s *xlStorage) GetDiskLoc() (poolIdx, setIdx, diskIdx int) {
	s.RLock()
	defer s.RUnlock()
	// If unset, see if we can locate it.
	if s.poolIndex < 0 || s.setIndex < 0 || s.diskIndex < 0 {
		return getXLDiskLoc(s.diskID)
	}
	return s.poolIndex, s.setIndex, s.diskIndex
}

// Set location indexes.
func (s *xlStorage) SetDiskLoc(poolIdx, setIdx, diskIdx int) {
	s.poolIndex = poolIdx
	s.setIndex = setIdx
	s.diskIndex = diskIdx
}

func (s *xlStorage) Healing() *healingTracker {
	healingFile := pathJoin(s.diskPath, minioMetaBucket,
		bucketMetaPrefix, healingTrackerFilename)
	b, err := ioutil.ReadFile(healingFile)
	if err != nil {
		return nil
	}
	var h healingTracker
	_, err = h.UnmarshalMsg(b)
	logger.LogIf(GlobalContext, err)
	return &h
}

func (s *xlStorage) readMetadata(itemPath string) ([]byte, error) {
	if err := checkPathLength(itemPath); err != nil {
		return nil, err
	}

	f, err := OpenFile(itemPath, readMode, 0)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	stat, err := f.Stat()
	if err != nil {
		return nil, err
	}
	if stat.IsDir() {
		return nil, &os.PathError{
			Op:   "open",
			Path: itemPath,
			Err:  syscall.EISDIR,
		}
	}
	return readXLMetaNoData(f, stat.Size())
}

func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry) (dataUsageCache, error) {
	// Updates must be closed before we return.
	defer close(updates)
	var lc *lifecycle.Lifecycle
	var err error

	// Check if the current bucket has a configured lifecycle policy
	if globalLifecycleSys != nil {
		lc, err = globalLifecycleSys.Get(cache.Info.Name)
		if err == nil && lc.HasActiveRules("", true) {
			cache.Info.lifeCycle = lc
			if intDataUpdateTracker.debug {
				console.Debugln(color.Green("scannerDisk:") + " lifecycle: Active rules found")
			}
		}
	}

	// Check if the current bucket has replication configuration
	if rcfg, err := globalBucketMetadataSys.GetReplicationConfig(ctx, cache.Info.Name); err == nil {
		if rcfg.HasActiveRules("", true) {
			tgt := globalBucketTargetSys.GetRemoteBucketTargetByArn(ctx, cache.Info.Name, rcfg.RoleArn)
			cache.Info.replication = replicationConfig{
				Config:          rcfg,
				ResetID:         tgt.ResetID,
				ResetBeforeDate: tgt.ResetBeforeDate}
			if intDataUpdateTracker.debug {
				console.Debugln(color.Green("scannerDisk:") + " replication: Active rules found")
			}
		}
	}
	// return initialized object layer
	objAPI := newObjectLayerFn()
	// object layer not initialized, return.
	if objAPI == nil {
		return cache, errServerNotInitialized
	}

	cache.Info.updates = updates

	dataUsageInfo, err := scanDataFolder(ctx, s.diskPath, cache, func(item scannerItem) (sizeSummary, error) {
		// Look for `xl.meta/xl.json' at the leaf.
		if !strings.HasSuffix(item.Path, SlashSeparator+xlStorageFormatFile) &&
			!strings.HasSuffix(item.Path, SlashSeparator+xlStorageFormatFileV1) {
			// if no xl.meta/xl.json found, skip the file.
			return sizeSummary{}, errSkipFile
		}

		buf, err := s.readMetadata(item.Path)
		if err != nil {
			if intDataUpdateTracker.debug {
				console.Debugf(color.Green("scannerBucket:")+" object path missing: %v: %w\n", item.Path, err)
			}
			return sizeSummary{}, errSkipFile
		}
		defer metaDataPoolPut(buf)

		// Remove filename which is the meta file.
		item.transformMetaDir()

		fivs, err := getAllFileInfoVersions(buf, item.bucket, item.objectPath())
		if err != nil {
			if intDataUpdateTracker.debug {
				console.Debugf(color.Green("scannerBucket:")+" reading xl.meta failed: %v: %w\n", item.Path, err)
			}
			return sizeSummary{}, errSkipFile
		}
		sizeS := sizeSummary{}
		for _, version := range fivs.Versions {
			oi := version.ToObjectInfo(item.bucket, item.objectPath())
			sz := item.applyActions(ctx, objAPI, oi, &sizeS)
			if !oi.DeleteMarker && sz == oi.Size {
				sizeS.versions++
			}
			sizeS.totalSize += sz
		}
		return sizeS, nil
	})

	if err != nil {
		return dataUsageInfo, err
	}

	dataUsageInfo.Info.LastUpdate = time.Now()
	return dataUsageInfo, nil
}

// DiskInfo provides current information about disk space usage,
// total free inodes and underlying filesystem.
func (s *xlStorage) DiskInfo(context.Context) (info DiskInfo, err error) {
	s.diskInfoCache.Once.Do(func() {
		s.diskInfoCache.TTL = time.Second
		s.diskInfoCache.Update = func() (interface{}, error) {
			dcinfo := DiskInfo{
				RootDisk:  s.rootDisk,
				MountPath: s.diskPath,
				Endpoint:  s.endpoint.String(),
			}
			di, err := getDiskInfo(s.diskPath)
			if err != nil {
				return dcinfo, err
			}
			dcinfo.Total = di.Total
			dcinfo.Free = di.Free
			dcinfo.Used = di.Used
			dcinfo.UsedInodes = di.Files - di.Ffree
			dcinfo.FreeInodes = di.Ffree
			dcinfo.FSType = di.FSType
			diskID, err := s.GetDiskID()
			if errors.Is(err, errUnformattedDisk) {
				// if we found an unformatted disk then
				// healing is automatically true.
				dcinfo.Healing = true
			} else {
				// Check if the disk is being healed if GetDiskID
				// returned any error other than fresh disk
				dcinfo.Healing = s.Healing() != nil
			}
			dcinfo.ID = diskID
			return dcinfo, err
		}
	})

	v, err := s.diskInfoCache.Get()
	info = v.(DiskInfo)
	return info, err
}

// getVolDir - will convert incoming volume names to
// corresponding valid volume names on the backend in a platform
// compatible way for all operating systems. If volume is not found
// an error is generated.
func (s *xlStorage) getVolDir(volume string) (string, error) {
	if volume == "" || volume == "." || volume == ".." {
		return "", errVolumeNotFound
	}
	volumeDir := pathJoin(s.diskPath, volume)
	return volumeDir, nil
}

func (s *xlStorage) checkFormatJSON() (os.FileInfo, error) {
	formatFile := pathJoin(s.diskPath, minioMetaBucket, formatConfigFile)
	fi, err := Lstat(formatFile)
	if err != nil {
		// If the disk is still not initialized.
		if osIsNotExist(err) {
			if err = Access(s.diskPath); err == nil {
				// Disk is present but missing `format.json`
				return nil, errUnformattedDisk
			}
			if osIsNotExist(err) {
				return nil, errDiskNotFound
			} else if osIsPermission(err) {
				return nil, errDiskAccessDenied
			}
			logger.LogIf(GlobalContext, err) // log unexpected errors
			return nil, errCorruptedFormat
		} else if osIsPermission(err) {
			return nil, errDiskAccessDenied
		}
		logger.LogIf(GlobalContext, err) // log unexpected errors
		return nil, errCorruptedFormat
	}
	return fi, nil
}

// GetDiskID - returns the cached disk uuid
func (s *xlStorage) GetDiskID() (string, error) {
	s.RLock()
	diskID := s.diskID
	fileInfo := s.formatFileInfo
	lastCheck := s.formatLastCheck

	// check if we have a valid disk ID that is less than 1 second old.
	if fileInfo != nil && diskID != "" && time.Since(lastCheck) <= time.Second {
		s.RUnlock()
		return diskID, nil
	}
	s.RUnlock()

	fi, err := s.checkFormatJSON()
	if err != nil {
		return "", err
	}

	if xioutil.SameFile(fi, fileInfo) && diskID != "" {
		s.Lock()
		// If the file has not changed, just return the cached diskID information.
		s.formatLastCheck = time.Now()
		s.Unlock()
		return diskID, nil
	}

	formatFile := pathJoin(s.diskPath, minioMetaBucket, formatConfigFile)
	b, err := xioutil.ReadFile(formatFile)
	if err != nil {
		// If the disk is still not initialized.
		if osIsNotExist(err) {
			if err = Access(s.diskPath); err == nil {
				// Disk is present but missing `format.json`
				return "", errUnformattedDisk
			}
			if osIsNotExist(err) {
				return "", errDiskNotFound
			} else if osIsPermission(err) {
				return "", errDiskAccessDenied
			}
			logger.LogIf(GlobalContext, err) // log unexpected errors
			return "", errCorruptedFormat
		} else if osIsPermission(err) {
			return "", errDiskAccessDenied
		}
		logger.LogIf(GlobalContext, err) // log unexpected errors
		return "", errCorruptedFormat
	}

	format := &formatErasureV3{}
	var json = jsoniter.ConfigCompatibleWithStandardLibrary
	if err = json.Unmarshal(b, &format); err != nil {
		logger.LogIf(GlobalContext, err) // log unexpected errors
		return "", errCorruptedFormat
	}

	s.Lock()
	defer s.Unlock()
	s.diskID = format.Erasure.This
	s.formatLegacy = format.Erasure.DistributionAlgo == formatErasureVersionV2DistributionAlgoV1
	s.formatFileInfo = fi
	s.formatLastCheck = time.Now()
	return s.diskID, nil
}

// Make a volume entry.
func (s *xlStorage) SetDiskID(id string) {
	// NO-OP for xlStorage as it is handled either by xlStorageDiskIDCheck{} for local disks or
	// storage rest server for remote disks.
}

func (s *xlStorage) MakeVolBulk(ctx context.Context, volumes ...string) error {
	for _, volume := range volumes {
		if err := s.MakeVol(ctx, volume); err != nil {
			if errors.Is(err, errDiskAccessDenied) {
				return errDiskAccessDenied
			}
		}
	}
	return nil
}

// Make a volume entry.
func (s *xlStorage) MakeVol(ctx context.Context, volume string) error {
	if !isValidVolname(volume) {
		return errInvalidArgument
	}

	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	if err = Access(volumeDir); err != nil {
		// Volume does not exist we proceed to create.
		if osIsNotExist(err) {
			// Make a volume entry, with mode 0777 mkdir honors system umask.
			err = mkdirAll(volumeDir, 0777)
		}
		if osIsPermission(err) {
			return errDiskAccessDenied
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	// Stat succeeds we return errVolumeExists.
	return errVolumeExists
}

// ListVols - list volumes.
func (s *xlStorage) ListVols(context.Context) (volsInfo []VolInfo, err error) {
	return listVols(s.diskPath)
}

// List all the volumes from diskPath.
func listVols(dirPath string) ([]VolInfo, error) {
	if err := checkPathLength(dirPath); err != nil {
		return nil, err
	}
	entries, err := readDir(dirPath)
	if err != nil {
		return nil, errDiskNotFound
	}
	volsInfo := make([]VolInfo, 0, len(entries))
	for _, entry := range entries {
		if !HasSuffix(entry, SlashSeparator) || !isValidVolname(pathutil.Clean(entry)) {
			// Skip if entry is neither a directory not a valid volume name.
			continue
		}
		volsInfo = append(volsInfo, VolInfo{
			Name: pathutil.Clean(entry),
		})
	}
	return volsInfo, nil
}

// StatVol - get volume info.
func (s *xlStorage) StatVol(ctx context.Context, volume string) (vol VolInfo, err error) {
	// Verify if volume is valid and it exists.
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return VolInfo{}, err
	}

	// Stat a volume entry.
	var st os.FileInfo
	st, err = Lstat(volumeDir)
	if err != nil {
		switch {
		case osIsNotExist(err):
			return VolInfo{}, errVolumeNotFound
		case osIsPermission(err):
			return VolInfo{}, errDiskAccessDenied
		case isSysErrIO(err):
			return VolInfo{}, errFaultyDisk
		default:
			return VolInfo{}, err
		}
	}
	// As os.Lstat() doesn't carry other than ModTime(), use ModTime()
	// as CreatedTime.
	createdTime := st.ModTime()
	return VolInfo{
		Name:    volume,
		Created: createdTime,
	}, nil
}

// DeleteVol - delete a volume.
func (s *xlStorage) DeleteVol(ctx context.Context, volume string, forceDelete bool) (err error) {
	// Verify if volume is valid and it exists.
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	if forceDelete {
		err = renameAll(volumeDir, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID()))
	} else {
		err = Remove(volumeDir)
	}

	if err != nil {
		switch {
		case osIsNotExist(err):
			return errVolumeNotFound
		case isSysErrNotEmpty(err):
			return errVolumeNotEmpty
		case osIsPermission(err):
			return errDiskAccessDenied
		case isSysErrIO(err):
			return errFaultyDisk
		default:
			return err
		}
	}
	return nil
}

// ListDir - return all the entries at the given directory path.
// If an entry is a directory it will be returned with a trailing SlashSeparator.
func (s *xlStorage) ListDir(ctx context.Context, volume, dirPath string, count int) (entries []string, err error) {
	// Verify if volume is valid and it exists.
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return nil, err
	}

	dirPathAbs := pathJoin(volumeDir, dirPath)
	if count > 0 {
		entries, err = readDirN(dirPathAbs, count)
	} else {
		entries, err = readDir(dirPathAbs)
	}
	if err != nil {
		if err == errFileNotFound {
			if ierr := Access(volumeDir); ierr != nil {
				if osIsNotExist(ierr) {
					return nil, errVolumeNotFound
				} else if isSysErrIO(ierr) {
					return nil, errFaultyDisk
				}
			}
		}
		return nil, err
	}

	return entries, nil
}

// DeleteVersions deletes slice of versions, it can be same object
// or multiple objects.
func (s *xlStorage) DeleteVersions(ctx context.Context, volume string, versions []FileInfo) []error {
	errs := make([]error, len(versions))

	for i, version := range versions {
		if err := s.DeleteVersion(ctx, volume, version.Name, version, false); err != nil {
			errs[i] = err
		}
	}

	return errs
}

// DeleteVersion - deletes FileInfo metadata for path at `xl.meta`. forceDelMarker
// will force creating a new `xl.meta` to create a new delete marker
func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool) error {
	if HasSuffix(path, SlashSeparator) {
		return s.Delete(ctx, volume, path, false)
	}

	buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
	if err != nil {
		if err != errFileNotFound {
			return err
		}
		if fi.Deleted && forceDelMarker {
			// Create a new xl.meta with a delete marker in it
			return s.WriteMetadata(ctx, volume, path, fi)
		}
		if fi.VersionID != "" {
			return errFileVersionNotFound
		}
		return errFileNotFound
	}

	if len(buf) == 0 {
		if fi.VersionID != "" {
			return errFileVersionNotFound
		}
		return errFileNotFound
	}

	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	if !isXL2V1Format(buf) {
		// Delete the meta file, if there are no more versions the
		// top level parent is automatically removed.
		return s.deleteFile(volumeDir, pathJoin(volumeDir, path), true)
	}

	var xlMeta xlMetaV2
	if err = xlMeta.Load(buf); err != nil {
		return err
	}

	dataDir, lastVersion, err := xlMeta.DeleteVersion(fi)
	if err != nil {
		return err
	}
	if dataDir != "" {
		versionID := fi.VersionID
		if versionID == "" {
			versionID = nullVersionID
		}
		// PR #11758 used DataDir, preserve it
		// for users who might have used master
		// branch
		if !xlMeta.data.remove(versionID, dataDir) {
			filePath := pathJoin(volumeDir, path, dataDir)
			if err = checkPathLength(filePath); err != nil {
				return err
			}
			if err = renameAll(filePath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID())); err != nil {
				if err != errFileNotFound {
					return err
				}
			}
		}
	}
	if !lastVersion {
		buf, err = xlMeta.AppendTo(metaDataPoolGet())
		defer metaDataPoolPut(buf)
		if err != nil {
			return err
		}

		return s.WriteAll(ctx, volume, pathJoin(path, xlStorageFormatFile), buf)
	}

	// Move xl.meta to trash
	filePath := pathJoin(volumeDir, path, xlStorageFormatFile)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	err = Rename(filePath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID()))
	if err == nil || err == errFileNotFound {
		s.deleteFile(volumeDir, pathJoin(volumeDir, path), false)
	}
	return err
}

// Updates only metadata for a given version.
func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo) error {
	if len(fi.Metadata) == 0 {
		return errInvalidArgument
	}

	buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
	if err != nil {
		if err == errFileNotFound {
			if fi.VersionID != "" {
				return errFileVersionNotFound
			}
		}
		return err
	}

	if !isXL2V1Format(buf) {
		return errFileVersionNotFound
	}

	var xlMeta xlMetaV2
	if err = xlMeta.Load(buf); err != nil {
		logger.LogIf(ctx, err)
		return err
	}

	if err = xlMeta.UpdateObjectVersion(fi); err != nil {
		return err
	}

	buf, err = xlMeta.AppendTo(nil)
	if err != nil {
		return err
	}

	return s.WriteAll(ctx, volume, pathJoin(path, xlStorageFormatFile), buf)
}

// WriteMetadata - writes FileInfo metadata for path at `xl.meta`
func (s *xlStorage) WriteMetadata(ctx context.Context, volume, path string, fi FileInfo) error {
	if fi.Fresh {
		var xlMeta xlMetaV2
		if err := xlMeta.AddVersion(fi); err != nil {
			logger.LogIf(ctx, err)
			return err
		}
		buf, err := xlMeta.AppendTo(metaDataPoolGet())
		defer metaDataPoolPut(buf)
		if err != nil {
			logger.LogIf(ctx, err)
			return err
		}
		// First writes for special situations do not write to stable storage.
		// this is currently used by
		// - emphemeral objects such as objects created during listObjects() calls
		// - newMultipartUpload() call..
		return s.writeAll(ctx, volume, pathJoin(path, xlStorageFormatFile), buf, false)
	}

	buf, err := s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
	if err != nil && err != errFileNotFound {
		return err
	}
	defer metaDataPoolPut(buf)

	var xlMeta xlMetaV2
	if !isXL2V1Format(buf) {
		err = xlMeta.AddVersion(fi)
		if err != nil {
			logger.LogIf(ctx, err)
			return err
		}

		buf, err = xlMeta.AppendTo(metaDataPoolGet())
		defer metaDataPoolPut(buf)
		if err != nil {
			logger.LogIf(ctx, err)
			return err
		}
	} else {
		if err = xlMeta.Load(buf); err != nil {
			logger.LogIf(ctx, err)
			return err
		}

		if err = xlMeta.AddVersion(fi); err != nil {
			logger.LogIf(ctx, err)
			return err
		}

		buf, err = xlMeta.AppendTo(metaDataPoolGet())
		defer metaDataPoolPut(buf)
		if err != nil {
			logger.LogIf(ctx, err)
			return err
		}
	}

	return s.WriteAll(ctx, volume, pathJoin(path, xlStorageFormatFile), buf)
}

func (s *xlStorage) renameLegacyMetadata(volumeDir, path string) (err error) {
	s.RLock()
	legacy := s.formatLegacy
	s.RUnlock()
	if !legacy {
		// if its not a legacy backend then this function is
		// a no-op always returns errFileNotFound
		return errFileNotFound
	}

	// Validate file path length, before reading.
	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	srcFilePath := pathJoin(filePath, xlStorageFormatFileV1)
	dstFilePath := pathJoin(filePath, xlStorageFormatFile)

	// Renaming xl.json to xl.meta should be fully synced to disk.
	defer func() {
		if err == nil {
			if s.globalSync {
				// Sync to disk only upon success.
				globalSync()
			}
		}
	}()

	if err = Rename(srcFilePath, dstFilePath); err != nil {
		switch {
		case isSysErrNotDir(err):
			return errFileNotFound
		case isSysErrPathNotFound(err):
			return errFileNotFound
		case isSysErrCrossDevice(err):
			return fmt.Errorf("%w (%s)->(%s)", errCrossDeviceLink, srcFilePath, dstFilePath)
		case osIsNotExist(err):
			return errFileNotFound
		case osIsExist(err):
			// This is returned only when destination is a directory and we
			// are attempting a rename from file to directory.
			return errIsNotRegular
		default:
			return err
		}
	}
	return nil
}

// ReadVersion - reads metadata and returns FileInfo at path `xl.meta`
// for all objects less than `32KiB` this call returns data as well
// along with metadata.
func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID string, readData bool) (fi FileInfo, err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return fi, err
	}
	var buf []byte
	if readData {
		buf, err = s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
	} else {
		buf, err = s.readMetadata(pathJoin(volumeDir, path, xlStorageFormatFile))
		if err != nil {
			if osIsNotExist(err) {
				if err = Access(volumeDir); err != nil && osIsNotExist(err) {
					return fi, errVolumeNotFound
				}
			}
			err = osErrToFileErr(err)
		}
	}

	if err != nil {
		if err == errFileNotFound {
			if err = s.renameLegacyMetadata(volumeDir, path); err != nil {
				if err == errFileNotFound {
					if versionID != "" {
						return fi, errFileVersionNotFound
					}
					return fi, errFileNotFound
				}
				return fi, err
			}
			buf, err = s.ReadAll(ctx, volume, pathJoin(path, xlStorageFormatFile))
			if err != nil {
				if err == errFileNotFound {
					if versionID != "" {
						return fi, errFileVersionNotFound
					}
					return fi, errFileNotFound
				}
				return fi, err
			}
		} else {
			return fi, err
		}
	}

	if len(buf) == 0 {
		if versionID != "" {
			return fi, errFileVersionNotFound
		}
		return fi, errFileNotFound
	}

	fi, err = getFileInfo(buf, volume, path, versionID, readData)
	if err != nil {
		return fi, err
	}

	if len(fi.Data) == 0 {
		// We did not read inline data, so we have no references.
		defer metaDataPoolPut(buf)
	}

	if readData {
		if len(fi.Data) > 0 || fi.Size == 0 {
			if fi.InlineData() {
				// If written with header we are fine.
				return fi, nil
			}
			if fi.Size == 0 || !(fi.VersionID != "" && fi.VersionID != nullVersionID) {
				// If versioned we have no conflicts.
				fi.SetInlineData()
				return fi, nil
			}

			// For overwritten objects without header we might have a conflict with
			// data written later.
			// Check the data path if there is a part with data.
			partPath := fmt.Sprintf("part.%d", fi.Parts[0].Number)
			dataPath := pathJoin(path, fi.DataDir, partPath)
			_, err = s.StatInfoFile(ctx, volume, dataPath)
			if err != nil {
				// Set the inline header, our inlined data is fine.
				fi.SetInlineData()
				return fi, nil
			}
			// Data exists on disk, remove the version from metadata.
			fi.Data = nil
		}

		// Reading data for small objects when
		// - object has not yet transitioned
		// - object size lesser than 128KiB
		// - object has maximum of 1 parts
		if fi.TransitionStatus == "" &&
			fi.DataDir != "" && fi.Size <= smallFileThreshold &&
			len(fi.Parts) == 1 {
			partPath := fmt.Sprintf("part.%d", fi.Parts[0].Number)
			dataPath := pathJoin(volumeDir, path, fi.DataDir, partPath)
			fi.Data, err = s.readAllData(volumeDir, dataPath)
			if err != nil {
				return FileInfo{}, err
			}
		}
	}

	return fi, nil
}

func (s *xlStorage) readAllData(volumeDir string, filePath string) (buf []byte, err error) {
	f, err := OpenFileDirectIO(filePath, readMode, 0666)
	if err != nil {
		if osIsNotExist(err) {
			// Check if the object doesn't exist because its bucket
			// is missing in order to return the correct error.
			if err = Access(volumeDir); err != nil && osIsNotExist(err) {
				return nil, errVolumeNotFound
			}
			return nil, errFileNotFound
		} else if osIsPermission(err) {
			return nil, errFileAccessDenied
		} else if isSysErrNotDir(err) || isSysErrIsDir(err) {
			return nil, errFileNotFound
		} else if isSysErrHandleInvalid(err) {
			// This case is special and needs to be handled for windows.
			return nil, errFileNotFound
		} else if isSysErrIO(err) {
			return nil, errFaultyDisk
		} else if isSysErrTooManyFiles(err) {
			return nil, errTooManyOpenFiles
		} else if isSysErrInvalidArg(err) {
			st, _ := Lstat(filePath)
			if st != nil && st.IsDir() {
				// Linux returns InvalidArg for directory O_DIRECT
				// we need to keep this fallback code to return correct
				// errors upwards.
				return nil, errFileNotFound
			}
			return nil, errUnsupportedDisk
		}
		return nil, err
	}
	r := &odirectReader{f, nil, nil, true, true, s, nil}
	defer r.Close()
	buf, err = ioutil.ReadAll(r)
	if err != nil {
		err = osErrToFileErr(err)
	}

	return buf, err
}

// ReadAll reads from r until an error or EOF and returns the data it read.
// A successful call returns err == nil, not err == EOF. Because ReadAll is
// defined to read from src until EOF, it does not treat an EOF from Read
// as an error to be reported.
// This API is meant to be used on files which have small memory footprint, do
// not use this on large files as it would cause server to crash.
func (s *xlStorage) ReadAll(ctx context.Context, volume string, path string) (buf []byte, err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return nil, err
	}

	// Validate file path length, before reading.
	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return nil, err
	}

	return s.readAllData(volumeDir, filePath)
}

// ReadFile reads exactly len(buf) bytes into buf. It returns the
// number of bytes copied. The error is EOF only if no bytes were
// read. On return, n == len(buf) if and only if err == nil. n == 0
// for io.EOF.
//
// If an EOF happens after reading some but not all the bytes,
// ReadFile returns ErrUnexpectedEOF.
//
// If the BitrotVerifier is not nil or not verified ReadFile
// tries to verify whether the disk has bitrot.
//
// Additionally ReadFile also starts reading from an offset. ReadFile
// semantics are same as io.ReadFull.
func (s *xlStorage) ReadFile(ctx context.Context, volume string, path string, offset int64, buffer []byte, verifier *BitrotVerifier) (int64, error) {
	if offset < 0 {
		return 0, errInvalidArgument
	}

	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return 0, err
	}

	var n int

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return 0, errVolumeNotFound
		} else if isSysErrIO(err) {
			return 0, errFaultyDisk
		} else if osIsPermission(err) {
			return 0, errFileAccessDenied
		}
		return 0, err
	}

	// Validate effective path length before reading.
	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return 0, err
	}

	// Open the file for reading.
	file, err := Open(filePath)
	if err != nil {
		switch {
		case osIsNotExist(err):
			return 0, errFileNotFound
		case osIsPermission(err):
			return 0, errFileAccessDenied
		case isSysErrNotDir(err):
			return 0, errFileAccessDenied
		case isSysErrIO(err):
			return 0, errFaultyDisk
		case isSysErrTooManyFiles(err):
			return 0, errTooManyOpenFiles
		default:
			return 0, err
		}
	}

	// Close the file descriptor.
	defer file.Close()

	st, err := file.Stat()
	if err != nil {
		return 0, err
	}

	// Verify it is a regular file, otherwise subsequent Seek is
	// undefined.
	if !st.Mode().IsRegular() {
		return 0, errIsNotRegular
	}

	if verifier == nil {
		n, err = file.ReadAt(buffer, offset)
		return int64(n), err
	}

	h := verifier.algorithm.New()
	if _, err = io.Copy(h, io.LimitReader(file, offset)); err != nil {
		return 0, err
	}

	if n, err = io.ReadFull(file, buffer); err != nil {
		return int64(n), err
	}

	if _, err = h.Write(buffer); err != nil {
		return 0, err
	}

	if _, err = io.Copy(h, file); err != nil {
		return 0, err
	}

	if !bytes.Equal(h.Sum(nil), verifier.sum) {
		return 0, errFileCorrupt
	}

	return int64(len(buffer)), nil
}

func (s *xlStorage) openFileSync(filePath string, mode int) (f *os.File, err error) {
	// Create top level directories if they don't exist.
	// with mode 0777 mkdir honors system umask.
	if err = mkdirAll(pathutil.Dir(filePath), 0777); err != nil {
		return nil, osErrToFileErr(err)
	}

	w, err := OpenFile(filePath, mode|writeMode, 0666)
	if err != nil {
		// File path cannot be verified since one of the parents is a file.
		switch {
		case isSysErrIsDir(err):
			return nil, errIsNotRegular
		case osIsPermission(err):
			return nil, errFileAccessDenied
		case isSysErrIO(err):
			return nil, errFaultyDisk
		case isSysErrTooManyFiles(err):
			return nil, errTooManyOpenFiles
		default:
			return nil, err
		}
	}

	return w, nil
}

func (s *xlStorage) openFileNoSync(filePath string, mode int) (f *os.File, err error) {
	// Create top level directories if they don't exist.
	// with mode 0777 mkdir honors system umask.
	if err = mkdirAll(pathutil.Dir(filePath), 0777); err != nil {
		return nil, osErrToFileErr(err)
	}

	w, err := OpenFile(filePath, mode, 0666)
	if err != nil {
		// File path cannot be verified since one of the parents is a file.
		switch {
		case isSysErrIsDir(err):
			return nil, errIsNotRegular
		case osIsPermission(err):
			return nil, errFileAccessDenied
		case isSysErrIO(err):
			return nil, errFaultyDisk
		case isSysErrTooManyFiles(err):
			return nil, errTooManyOpenFiles
		default:
			return nil, err
		}
	}

	return w, nil
}

// To support O_DIRECT reads for erasure backends.
type odirectReader struct {
	f         *os.File
	buf       []byte
	bufp      *[]byte
	freshRead bool
	smallFile bool
	s         *xlStorage
	err       error
}

// Read - Implements Reader interface.
func (o *odirectReader) Read(buf []byte) (n int, err error) {
	if o.err != nil && (len(o.buf) == 0 || o.freshRead) {
		return 0, o.err
	}
	if o.buf == nil {
		if o.smallFile {
			o.bufp = xlPoolSmall.Get().(*[]byte)
		} else {
			o.bufp = xlPoolLarge.Get().(*[]byte)
		}
	}
	if o.freshRead {
		o.buf = *o.bufp
		n, err = o.f.Read(o.buf)
		if err != nil && err != io.EOF {
			if isSysErrInvalidArg(err) {
				if err = disk.DisableDirectIO(o.f); err != nil {
					o.err = err
					return n, err
				}
				n, err = o.f.Read(o.buf)
			}
			if err != nil && err != io.EOF {
				o.err = err
				return n, err
			}
		}
		if n == 0 {
			// err is likely io.EOF
			o.err = err
			return n, err
		}
		o.err = err
		o.buf = o.buf[:n]
		o.freshRead = false
	}
	if len(buf) >= len(o.buf) {
		n = copy(buf, o.buf)
		o.freshRead = true
		return n, o.err
	}
	n = copy(buf, o.buf)
	o.buf = o.buf[n:]
	// There is more left in buffer, do not return any EOF yet.
	return n, nil
}

// Close - Release the buffer and close the file.
func (o *odirectReader) Close() error {
	if o.smallFile {
		xlPoolSmall.Put(o.bufp)
	} else {
		xlPoolLarge.Put(o.bufp)
	}
	return o.f.Close()
}

// ReadFileStream - Returns the read stream of the file.
func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, offset, length int64) (io.ReadCloser, error) {
	if offset < 0 {
		return nil, errInvalidArgument
	}

	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return nil, err
	}

	// Validate effective path length before reading.
	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return nil, err
	}

	var file *os.File
	// O_DIRECT only supported if offset is zero
	if offset == 0 {
		file, err = OpenFileDirectIO(filePath, readMode, 0666)
	} else {
		// Open the file for reading.
		file, err = OpenFile(filePath, readMode, 0666)
	}
	if err != nil {
		switch {
		case osIsNotExist(err):
			if err = Access(volumeDir); err != nil && osIsNotExist(err) {
				return nil, errVolumeNotFound
			}
			return nil, errFileNotFound
		case osIsPermission(err):
			return nil, errFileAccessDenied
		case isSysErrNotDir(err):
			return nil, errFileAccessDenied
		case isSysErrIO(err):
			return nil, errFaultyDisk
		case isSysErrTooManyFiles(err):
			return nil, errTooManyOpenFiles
		case isSysErrInvalidArg(err):
			return nil, errUnsupportedDisk
		default:
			return nil, err
		}
	}

	st, err := file.Stat()
	if err != nil {
		file.Close()
		return nil, err
	}

	// Verify it is a regular file, otherwise subsequent Seek is
	// undefined.
	if !st.Mode().IsRegular() {
		file.Close()
		return nil, errIsNotRegular
	}

	if offset == 0 {
		or := &odirectReader{file, nil, nil, true, false, s, nil}
		if length <= smallFileThreshold {
			or = &odirectReader{file, nil, nil, true, true, s, nil}
		}
		r := struct {
			io.Reader
			io.Closer
		}{Reader: io.LimitReader(or, length), Closer: closeWrapper(func() error {
			return or.Close()
		})}
		return r, nil
	}

	r := struct {
		io.Reader
		io.Closer
	}{Reader: io.LimitReader(file, length), Closer: closeWrapper(func() error {
		return file.Close()
	})}

	if offset > 0 {
		if _, err = file.Seek(offset, io.SeekStart); err != nil {
			r.Close()
			return nil, err
		}
	}

	// Add readahead to big reads
	if length >= readAheadSize {
		rc, err := readahead.NewReadCloserSize(r, readAheadBuffers, readAheadBufSize)
		if err != nil {
			r.Close()
			return nil, err
		}
		return rc, nil
	}

	// Just add a small 64k buffer.
	r.Reader = bufio.NewReaderSize(r.Reader, 64<<10)
	return r, nil
}

// closeWrapper converts a function to an io.Closer
type closeWrapper func() error

// Close calls the wrapped function.
func (c closeWrapper) Close() error {
	return c()
}

// CreateFile - creates the file.
func (s *xlStorage) CreateFile(ctx context.Context, volume, path string, fileSize int64, r io.Reader) (err error) {
	if fileSize < -1 {
		return errInvalidArgument
	}

	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	parentFilePath := pathutil.Dir(filePath)
	defer func() {
		if err != nil {
			if volume == minioMetaTmpBucket {
				// only cleanup parent path if the
				// parent volume name is minioMetaTmpBucket
				removeAll(parentFilePath)
			}
		}
	}()

	if fileSize >= 0 && fileSize <= smallFileThreshold {
		// For streams smaller than 128KiB we simply write them as O_DSYNC (fdatasync)
		// and not O_DIRECT to avoid the complexities of aligned I/O.
		w, err := s.openFileSync(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL)
		if err != nil {
			return err
		}
		defer w.Close()

		written, err := io.Copy(w, r)
		if err != nil {
			return osErrToFileErr(err)
		}

		if written < fileSize {
			return errLessData
		} else if written > fileSize {
			return errMoreData
		}

		return nil
	}

	// Create top level directories if they don't exist.
	// with mode 0777 mkdir honors system umask.
	if err = mkdirAll(parentFilePath, 0777); err != nil {
		return osErrToFileErr(err)
	}

	w, err := OpenFileDirectIO(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0666)
	if err != nil {
		return osErrToFileErr(err)
	}

	defer func() {
		disk.Fdatasync(w) // Only interested in flushing the size_t not mtime/atime
		w.Close()
	}()

	var bufp *[]byte
	if fileSize > 0 && fileSize >= reallyLargeFileThreshold {
		// use a larger 4MiB buffer for really large streams.
		bufp = xlPoolReallyLarge.Get().(*[]byte)
		defer xlPoolReallyLarge.Put(bufp)
	} else {
		bufp = xlPoolLarge.Get().(*[]byte)
		defer xlPoolLarge.Put(bufp)
	}

	written, err := xioutil.CopyAligned(w, r, *bufp, fileSize)
	if err != nil {
		return err
	}

	if written < fileSize && fileSize >= 0 {
		return errLessData
	} else if written > fileSize && fileSize >= 0 {
		return errMoreData
	}

	return nil
}

func (s *xlStorage) writeAll(ctx context.Context, volume string, path string, b []byte, sync bool) (err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	var w *os.File
	if sync {
		w, err = s.openFileSync(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC)
	} else {
		w, err = s.openFileNoSync(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC)
	}
	if err != nil {
		return err
	}
	defer w.Close()

	n, err := w.Write(b)
	if err != nil {
		return err
	}

	if n != len(b) {
		return io.ErrShortWrite
	}

	return nil
}

func (s *xlStorage) WriteAll(ctx context.Context, volume string, path string, b []byte) (err error) {
	return s.writeAll(ctx, volume, path, b, true)
}

// AppendFile - append a byte array at path, if file doesn't exist at
// path this call explicitly creates it.
func (s *xlStorage) AppendFile(ctx context.Context, volume string, path string, buf []byte) (err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if osIsPermission(err) {
			return errVolumeAccessDenied
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	var w *os.File
	// Create file if not found. Not doing O_DIRECT here to avoid the code that does buffer aligned writes.
	// AppendFile() is only used by healing code to heal objects written in old format.
	w, err = s.openFileSync(filePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY)
	if err != nil {
		return err
	}
	defer w.Close()

	n, err := w.Write(buf)
	if err != nil {
		return err
	}

	if n != len(buf) {
		return io.ErrShortWrite
	}

	return nil
}

// CheckParts check if path has necessary parts available.
func (s *xlStorage) CheckParts(ctx context.Context, volume string, path string, fi FileInfo) error {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		}
		return err
	}

	for _, part := range fi.Parts {
		partPath := pathJoin(path, fi.DataDir, fmt.Sprintf("part.%d", part.Number))
		filePath := pathJoin(volumeDir, partPath)
		if err = checkPathLength(filePath); err != nil {
			return err
		}
		st, err := Lstat(filePath)
		if err != nil {
			return osErrToFileErr(err)
		}
		if st.Mode().IsDir() {
			return errFileNotFound
		}
		// Check if shard is truncated.
		if st.Size() < fi.Erasure.ShardFileSize(part.Size) {
			return errFileCorrupt
		}
	}

	return nil
}

// deleteFile deletes a file or a directory if its empty unless recursive
// is set to true. If the target is successfully deleted, it will recursively
// move up the tree, deleting empty parent directories until it finds one
// with files in it. Returns nil for a non-empty directory even when
// recursive is set to false.
func (s *xlStorage) deleteFile(basePath, deletePath string, recursive bool) error {
	if basePath == "" || deletePath == "" {
		return nil
	}
	isObjectDir := HasSuffix(deletePath, SlashSeparator)
	basePath = pathutil.Clean(basePath)
	deletePath = pathutil.Clean(deletePath)
	if !strings.HasPrefix(deletePath, basePath) || deletePath == basePath {
		return nil
	}

	var err error
	if recursive {
		err = renameAll(deletePath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID()))
	} else {
		err = Remove(deletePath)
	}
	if err != nil {
		switch {
		case isSysErrNotEmpty(err):
			// if object is a directory, but if its not empty
			// return FileNotFound to indicate its an empty prefix.
			if isObjectDir {
				return errFileNotFound
			}
			// Ignore errors if the directory is not empty. The server relies on
			// this functionality, and sometimes uses recursion that should not
			// error on parent directories.
			return nil
		case osIsNotExist(err):
			return nil
		case osIsPermission(err):
			return errFileAccessDenied
		case isSysErrIO(err):
			return errFaultyDisk
		default:
			return err
		}
	}

	deletePath = pathutil.Dir(deletePath)

	// Delete parent directory obviously not recursively. Errors for
	// parent directories shouldn't trickle down.
	s.deleteFile(basePath, deletePath, false)

	return nil
}

// DeleteFile - delete a file at path.
func (s *xlStorage) Delete(ctx context.Context, volume string, path string, recursive bool) (err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if osIsPermission(err) {
			return errVolumeAccessDenied
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	// Following code is needed so that we retain SlashSeparator suffix if any in
	// path argument.
	filePath := pathJoin(volumeDir, path)
	if err = checkPathLength(filePath); err != nil {
		return err
	}

	// Delete file and delete parent directory as well if it's empty.
	return s.deleteFile(volumeDir, filePath, recursive)
}

// RenameData - rename source path to destination path atomically, metadata and data directory.
func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) (err error) {
	defer func() {
		if err == nil {
			if s.globalSync {
				globalSync()
			}
		}
	}()

	srcVolumeDir, err := s.getVolDir(srcVolume)
	if err != nil {
		return err
	}

	dstVolumeDir, err := s.getVolDir(dstVolume)
	if err != nil {
		return err
	}

	// Stat a volume entry.
	if err = Access(srcVolumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	if err = Access(dstVolumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	srcFilePath := pathutil.Join(srcVolumeDir, pathJoin(srcPath, xlStorageFormatFile))
	dstFilePath := pathutil.Join(dstVolumeDir, pathJoin(dstPath, xlStorageFormatFile))

	var srcDataPath string
	var dstDataPath string
	var dataDir string
	if !fi.IsRemote() {
		dataDir = retainSlash(fi.DataDir)
	}
	if dataDir != "" {
		srcDataPath = retainSlash(pathJoin(srcVolumeDir, srcPath, dataDir))
		// make sure to always use path.Join here, do not use pathJoin as
		// it would additionally add `/` at the end and it comes in the
		// way of renameAll(), parentDir creation.
		dstDataPath = pathutil.Join(dstVolumeDir, dstPath, dataDir)
	}

	if err = checkPathLength(srcFilePath); err != nil {
		return err
	}

	if err = checkPathLength(dstFilePath); err != nil {
		return err
	}

	dstBuf, err := xioutil.ReadFile(dstFilePath)
	if err != nil {
		// handle situations when dstFilePath is 'file'
		// for example such as someone is trying to
		// upload an object such as `prefix/object/xl.meta`
		// where `prefix/object` is already an object
		if isSysErrNotDir(err) && runtime.GOOS != globalWindowsOSName {
			// NOTE: On windows the error happens at
			// next line and returns appropriate error.
			return errFileAccessDenied
		}
		if !osIsNotExist(err) {
			return osErrToFileErr(err)
		}
		// errFileNotFound comes here.
		err = s.renameLegacyMetadata(dstVolumeDir, dstPath)
		if err != nil && err != errFileNotFound {
			return err
		}
		if err == nil {
			dstBuf, err = xioutil.ReadFile(dstFilePath)
			if err != nil && !osIsNotExist(err) {
				return osErrToFileErr(err)
			}
		}
	}

	var xlMeta xlMetaV2
	var legacyPreserved bool
	if len(dstBuf) > 0 {
		if isXL2V1Format(dstBuf) {
			if err = xlMeta.Load(dstBuf); err != nil {
				logger.LogIf(s.ctx, err)
				return err
			}
		} else {
			// This code-path is to preserve the legacy data.
			xlMetaLegacy := &xlMetaV1Object{}
			var json = jsoniter.ConfigCompatibleWithStandardLibrary
			if err := json.Unmarshal(dstBuf, xlMetaLegacy); err != nil {
				logger.LogIf(s.ctx, err)
				return errFileCorrupt
			}
			if err = xlMeta.AddLegacy(xlMetaLegacy); err != nil {
				logger.LogIf(s.ctx, err)
				return errFileCorrupt
			}
			legacyPreserved = true
		}
	} else {
		s.RLock()
		formatLegacy := s.formatLegacy
		s.RUnlock()
		// It is possible that some drives may not have `xl.meta` file
		// in such scenarios verify if atleast `part.1` files exist
		// to verify for legacy version.
		if formatLegacy {
			// We only need this code if we are moving
			// from `xl.json` to `xl.meta`, we can avoid
			// one extra readdir operation here for all
			// new deployments.
			currentDataPath := pathJoin(dstVolumeDir, dstPath)
			entries, err := readDirN(currentDataPath, 1)
			if err != nil && err != errFileNotFound {
				return osErrToFileErr(err)
			}
			for _, entry := range entries {
				if entry == xlStorageFormatFile || strings.HasSuffix(entry, slashSeparator) {
					continue
				}
				if strings.HasPrefix(entry, "part.") {
					legacyPreserved = true
					break
				}
			}
		}
	}

	legacyDataPath := pathJoin(dstVolumeDir, dstPath, legacyDataDir)
	if legacyPreserved {
		// Preserve all the legacy data, could be slow, but at max there can be 10,000 parts.
		currentDataPath := pathJoin(dstVolumeDir, dstPath)
		entries, err := readDir(currentDataPath)
		if err != nil {
			return osErrToFileErr(err)
		}

		// legacy data dir means its old content, honor system umask.
		if err = mkdirAll(legacyDataPath, 0777); err != nil {
			// any failed mkdir-calls delete them.
			s.deleteFile(dstVolumeDir, legacyDataPath, true)
			return osErrToFileErr(err)
		}

		for _, entry := range entries {
			// Skip xl.meta renames further, also ignore any directories such as `legacyDataDir`
			if entry == xlStorageFormatFile || strings.HasSuffix(entry, slashSeparator) {
				continue
			}

			if err = Rename(pathJoin(currentDataPath, entry), pathJoin(legacyDataPath, entry)); err != nil {
				// Any failed rename calls un-roll previous transaction.
				s.deleteFile(dstVolumeDir, legacyDataPath, true)

				return osErrToFileErr(err)
			}
		}
	}

	var oldDstDataPath string
	if fi.VersionID == "" {
		// return the latest "null" versionId info
		ofi, err := xlMeta.ToFileInfo(dstVolume, dstPath, nullVersionID)
		if err == nil && !ofi.Deleted {
			if xlMeta.SharedDataDirCountStr(nullVersionID, ofi.DataDir) == 0 {
				// Purge the destination path as we are not preserving anything
				// versioned object was not requested.
				oldDstDataPath = pathJoin(dstVolumeDir, dstPath, ofi.DataDir)
				// if old destination path is same as new destination path
				// there is nothing to purge, this is true in case of healing
				// avoid setting oldDstDataPath at that point.
				if oldDstDataPath == dstDataPath {
					oldDstDataPath = ""
				}
				xlMeta.data.remove(nullVersionID, ofi.DataDir)
			}
		}
		// Empty fi.VersionID indicates that versioning is either
		// suspended or disabled on this bucket. RenameData will replace
		// the 'null' version. We add a free-version to track its tiered
		// content for asynchronous deletion.
		xlMeta.AddFreeVersion(fi)
	}

	if err = xlMeta.AddVersion(fi); err != nil {
		if legacyPreserved {
			// Any failed rename calls un-roll previous transaction.
			s.deleteFile(dstVolumeDir, legacyDataPath, true)
		}
		return err
	}

	dstBuf, err = xlMeta.AppendTo(metaDataPoolGet())
	defer metaDataPoolPut(dstBuf)
	if err != nil {
		logger.LogIf(ctx, err)
		if legacyPreserved {
			// Any failed rename calls un-roll previous transaction.
			s.deleteFile(dstVolumeDir, legacyDataPath, true)
		}
		return errFileCorrupt
	}

	if srcDataPath != "" {
		if err = s.WriteAll(ctx, srcVolume, pathJoin(srcPath, xlStorageFormatFile), dstBuf); err != nil {
			if legacyPreserved {
				// Any failed rename calls un-roll previous transaction.
				s.deleteFile(dstVolumeDir, legacyDataPath, true)
			}
			return err
		}

		// renameAll only for objects that have xl.meta not saved inline.
		if len(fi.Data) == 0 && fi.Size > 0 {
			renameAll(dstDataPath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID()))
			if err = renameAll(srcDataPath, dstDataPath); err != nil {
				if legacyPreserved {
					// Any failed rename calls un-roll previous transaction.
					s.deleteFile(dstVolumeDir, legacyDataPath, true)
				}
				s.deleteFile(dstVolumeDir, dstDataPath, false)

				logger.LogIf(ctx, err)
				return osErrToFileErr(err)
			}
		}

		// Commit meta-file
		if err = renameAll(srcFilePath, dstFilePath); err != nil {
			if legacyPreserved {
				// Any failed rename calls un-roll previous transaction.
				s.deleteFile(dstVolumeDir, legacyDataPath, true)
			}
			s.deleteFile(dstVolumeDir, dstFilePath, false)

			logger.LogIf(ctx, err)
			return osErrToFileErr(err)
		}

		// additionally only purge older data at the end of the transaction of new data-dir
		// movement, this is to ensure that previous data references can co-exist for
		// any recoverability.
		if oldDstDataPath != "" {
			renameAll(oldDstDataPath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, mustGetUUID()))
		}
	} else {
		// Write meta-file directly, no data
		if err = s.WriteAll(ctx, dstVolume, pathJoin(dstPath, xlStorageFormatFile), dstBuf); err != nil {
			if legacyPreserved {
				// Any failed rename calls un-roll previous transaction.
				s.deleteFile(dstVolumeDir, legacyDataPath, true)
			}
			s.deleteFile(dstVolumeDir, dstFilePath, false)

			logger.LogIf(ctx, err)
			return err
		}
	}

	// srcFilePath is always in minioMetaTmpBucket, an attempt to
	// remove the temporary folder is enough since at this point
	// ideally all transaction should be complete.

	Remove(pathutil.Dir(srcFilePath))
	return nil
}

// RenameFile - rename source path to destination path atomically.
func (s *xlStorage) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string) (err error) {
	srcVolumeDir, err := s.getVolDir(srcVolume)
	if err != nil {
		return err
	}
	dstVolumeDir, err := s.getVolDir(dstVolume)
	if err != nil {
		return err
	}
	// Stat a volume entry.
	if err = Access(srcVolumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	if err = Access(dstVolumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if isSysErrIO(err) {
			return errFaultyDisk
		}
		return err
	}

	srcIsDir := HasSuffix(srcPath, SlashSeparator)
	dstIsDir := HasSuffix(dstPath, SlashSeparator)
	// Either src and dst have to be directories or files, else return error.
	if !(srcIsDir && dstIsDir || !srcIsDir && !dstIsDir) {
		return errFileAccessDenied
	}
	srcFilePath := pathutil.Join(srcVolumeDir, srcPath)
	if err = checkPathLength(srcFilePath); err != nil {
		return err
	}
	dstFilePath := pathutil.Join(dstVolumeDir, dstPath)
	if err = checkPathLength(dstFilePath); err != nil {
		return err
	}
	if srcIsDir {
		// If source is a directory, we expect the destination to be non-existent but we
		// we still need to allow overwriting an empty directory since it represents
		// an object empty directory.
		dirInfo, err := Lstat(dstFilePath)
		if isSysErrIO(err) {
			return errFaultyDisk
		}
		if err != nil {
			if !osIsNotExist(err) {
				return err
			}
		} else {
			if !dirInfo.IsDir() {
				return errFileAccessDenied
			}
			if err = Remove(dstFilePath); err != nil {
				if isSysErrNotEmpty(err) {
					return errFileAccessDenied
				}
				return err
			}
		}
	}

	if err = renameAll(srcFilePath, dstFilePath); err != nil {
		return osErrToFileErr(err)
	}

	// Remove parent dir of the source file if empty
	parentDir := pathutil.Dir(srcFilePath)
	s.deleteFile(srcVolumeDir, parentDir, false)

	return nil
}

func (s *xlStorage) bitrotVerify(partPath string, partSize int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
	// Open the file for reading.
	file, err := Open(partPath)
	if err != nil {
		return osErrToFileErr(err)
	}

	// Close the file descriptor.
	defer file.Close()
	fi, err := file.Stat()
	if err != nil {
		// Unable to stat on the file, return an expected error
		// for healing code to fix this file.
		return err
	}
	return bitrotVerify(file, fi.Size(), partSize, algo, sum, shardSize)
}

func (s *xlStorage) VerifyFile(ctx context.Context, volume, path string, fi FileInfo) (err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return err
	}

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return errVolumeNotFound
		} else if isSysErrIO(err) {
			return errFaultyDisk
		} else if osIsPermission(err) {
			return errVolumeAccessDenied
		}
		return err
	}

	erasure := fi.Erasure
	for _, part := range fi.Parts {
		checksumInfo := erasure.GetChecksumInfo(part.Number)
		partPath := pathJoin(volumeDir, path, fi.DataDir, fmt.Sprintf("part.%d", part.Number))
		if err := s.bitrotVerify(partPath,
			erasure.ShardFileSize(part.Size),
			checksumInfo.Algorithm,
			checksumInfo.Hash, erasure.ShardSize()); err != nil {
			if !IsErr(err, []error{
				errFileNotFound,
				errVolumeNotFound,
				errFileCorrupt,
			}...) {
				logger.GetReqInfo(s.ctx).AppendTags("disk", s.String())
				logger.LogIf(s.ctx, err)
			}
			return err
		}
	}

	return nil
}

func (s *xlStorage) StatInfoFile(ctx context.Context, volume, path string) (stat StatInfo, err error) {
	volumeDir, err := s.getVolDir(volume)
	if err != nil {
		return stat, err
	}

	// Stat a volume entry.
	if err = Access(volumeDir); err != nil {
		if osIsNotExist(err) {
			return stat, errVolumeNotFound
		} else if isSysErrIO(err) {
			return stat, errFaultyDisk
		} else if osIsPermission(err) {
			return stat, errVolumeAccessDenied
		}
		return stat, err
	}
	filePath := pathJoin(volumeDir, path)
	if err := checkPathLength(filePath); err != nil {
		return stat, err
	}
	st, _ := Lstat(filePath)
	if st == nil {
		return stat, errPathNotFound
	}

	return StatInfo{ModTime: st.ModTime(), Size: st.Size()}, nil
}


text:
Сидней (Сидни) Джон Хиксон (; 1859—1940) — английский ; член Лондонского королевского общества.

Биография

Сидни Джон Хиксон родился 25 июня 1859 году в Кембридже. Изучал медицину и естественные науки в университете и госпитале Святого Варфоломея в Лондоне, а затем в Кембридже.

По окончании обучения Хиксон преподавал сравнительную анатомию сначала в качестве лектора, а затем — профессора в Оксфордском университете, затем в университете родного города.

С 1894 года Сидней Джон Хиксон состоял профессором зоологии в Университете Манчестера.

С 1885 по 1886 год учёный путешествовал по Малайскому архипелагу для изучения коралловых рифов, а с 1889 по 1890 год работал в Канаде.

Хиксон поместил большое число печатных трудов зоологического содержания в изданиях: «Royal Society of Natural History», a кроме этого написал следующие книги: «А Naturalist in N. Celebes» (Лондон, 1889); «Fauna of the Deep Sea» (Лондон, 1894); «Story of Life, in the Seas» (Лондон, 1897).

Заслуги учёного были отмечены принятием его в члены Лондонского королевского общества.

Сидней Джон Хиксон умер 6 февраля 1940 года.

Примечания

Литература
 «Review: The Fauna of the Deep Sea by Sydney J. Hickson». Book News. vol. 12. 1894. p. 296.
 «Review: Story of Life in the Seas. The by Sydney J. Hickson». The Book News Monthly. vol. 21. 1903. p. 822.

Ссылки

Члены Лондонского королевского общества
Выпускники Кембриджского университета
Выпускники Даунинг-колледжа (Кембридж)

redpajama_set_name: RedPajamaWikipedia


text:
(TAP) – Tunisia is going through a difficult economic transition phase and hopes for its success, Prime Minster Youssef Chahed affirmed Thursday stressing that the second phase of the Tunisian democratic transition will only be done through the economic transition.
The National Unity Government's action plan is based on the Carthage Document that highlights development of renewable energies as a strategic choice to meet the upcoming challenges in the energy field and is one of the basic foundations to achieve sustainable development, preserve environment and ensure transfer of technology, he stressed at the opening of a conference on "Speeding up the Implementation of Renewable Energy Projects."
Sustainable industrial bases must also be set up especially given the increase of hydrocarbon's prices, decline in domestic natural gas resources and the start of the implementation of an action plan aiming to reduce greenhouse effects in accordance with the agreement of the 22nd session of the Conference of the Parties (COP 22).
The Premier recalled that Tunisia took the initiative years ago to promulgate provisions related to production of electricity from renewable energies and issued the appropriate application texts in addition to starting the achievement of renewable energies projects as part of a system of authorisations knowing that a set of projects will be announced as part of the concessions regime early 2018.
The implementation of these projects is part of the Tunisian Solar Plan (TSP)'s priorities that aims to achieve a set of objectives such as the production of 1,000 megawatts of renewable energies by 2020 that includes 650 photovoltaic energy and 350MW wind energy, i.e. 20% of the combined capacity estimated at about 2,500 million dinars, simultaneously with starting the first projects during the upcoming year (210MW).
The TSP aims to reduce consumption of conventional energy sources and the energy deficit, diversify the energy mix and limit the trade balance deficit which is mainly an energy deficit.
"The solar plan aims to reach a 12% rate of energy produced from renewable energies so as to reach 30% by 2030 in order to solidify natural energy resources," he said.
"Tunisia attaches a special importance to the implementation of the solar plan as part of its determination to provide energy and develop competitiveness of national economy in light of the changes of international prices of petroleum and their impacts on trade balance and public funding," he pointed out.

redpajama_set_name: RedPajamaCommonCrawl


If you live in Assembly Member Rob Bonta's district (he's on your ballot so you'll know) then these recommendations apply to you and your particular choices. Many of these choices overlap the assembly district but since a few don't, I'll stick to this part of the Central and East Oakland/Alameda/San Leandro electoral district.
First, remember that you can still register or reregister if you want to vote in the Democratic Primary, that is, in order to vote for Bernie, you have to be a Dem or a Decline to State but not a member of another party such as the American Independent Party-which is a right wing creation-not the designation for a non-party voter. You have until May 23rd!
Yes vote for Bernie if you want to see a real change in the party, at least at the top.But, in my opinion fighting for good government on the local level is the way we truly build change. The Republicans learned this years ago, and despite rumors of their demise, the old Mark Twain phrase, that party controls many statehouses, state legislatures, and the Congress by working their way up-not down. This is no small thing! It is why voters' rights and access to birth control are disappearing in many states and, affordable housing and transportation budgets have been whittled to nothing while student debt bankrupts the young.
So please join me in learning about local elections and get active in local politics. There's always more to do but it can get complicated. I'll start with the California Senate race in Senate District 9. Here we have the choice of two good candidates, three actually, including the almost unknown Katherine Welch but with just enough of a difference in vision to be important in our day-to-day lives-which is where we live them, after all.
Sandre Swanson, Oakland, Alameda, and San Leandro's previous assembly member was termed out and replaced by Rob Bonta-Rob is running for reelection-who, along with Congresswoman Barbara Lee-she's also on the ballot and the choice of over 80% of us-Assembly Member Tony Thurmond, and current State Senator Loni Hancock have endorsed him in his run for State Senate as Loni Hancock is now termed out.
Sandre bucked the established Democratic party and the governor who wanted to cut social programs like Healthy Children and eliminate Redevelopment-which the guv managed to do- and which has greatly exacerbated our housing crisis. Nancy has also done a lot of good work but we have to make a choice here–Skinner is well funded and has the support of Oakland Mayor Libby Schaaf which tells me something about their competing visions.
Sandre wrested local control back from the state for our school district, has fought to end the scourge of sexual slavery/trafficking of young girls in California, Oakland particularly, and established the Men and Boys of Color Commission which traveled the state gathering community info and putting it to use to create legislation to help these young men. There's more-Sandre is principled, progressive and will not back down to the party establishment. I'm voting for him.
I'm voting for Rob while pushing him to support tougher laws on police transparency and accountability, because I appreciate his hard work fighting for programs for children who need services that were rolled back by the governor. Both Rob and Tony Thurmond are working to bring benefits to marginalized communities but are often stymied by business Dems and a very frugal governor.
The big state race seems to be a foregone conclusion. Senator Barbara Boxer is retiring and despite the long list of candidates for US Senate, Kamala Harris is the favorite. She has all the right stuff, good connections, extensive credentials, a career as a top law enforcement officer and movie star looks-which we know count even if they shouldn't. I just have a little trouble consistently electing prosecutors who use the right lingo about criminal justice reform with very few specifics. I need specifics so I'm treading water on this race hoping to get some specific info thrown my way before proceeding to her anointment.
Keith Carson and Nate Miley are running for their supervisorial districts as incumbents. Both should get reelected. Carson is a prince among politicians who works quietly for the various constituencies in his district, listens to their needs and responds accordingly. Miley has his faults, one of them being his inability to deal with criticism. I have had my differences with him in the past but recognize the amount of time he puts in for his constituents and his penchant toward innovative solutions. His opponent has no experience relative to his ambition, believing he should start at the top without grappling with the contentious issues a supervisor faces.
Now to the judges. There are three judicial races which is an unusual number for one election. I can only suggest you follow the endorsements on their web pages if you have not attended any endorsement meetings. Since I have attended a few, I have a set of suggestions rather than recommended candidates.
Jennifer Madden who's running for Superior Court has been actively campaigning which cannot be said for Jonathon Van Ee. Jennifer has been endorsed by the Alameda Labor Council, the County Democratic Party and the John George Democratic Club, among others. It's true she is a prosecutor but she heads the Human Exploitation and Trafficking Unit (H.E.A.T.) and supervises H.E.A.T Watch, a 5 point strategy to combat Human Trafficking. I attended a panel discussion that BBBON, the Block by Block Organizing Network held a couple of years ago on how girls in Oakland are affected by this modern day slave trade and was impressed by her work. So I'm endorsing her. See https://maddenforjudge2016.com/ for more info. Vote Madden.
In the other contested Superior Court case, there seem to be some good candidates: Scott Jackson, David Lim and Barbara Thomas. I've seen Scott Jackson at three endorsement meetings and Lim and Thomas at one each.
The Alameda Labor Council has endorsed Lim, who as a council member in San Mateo helped google bus drivers unionize, while Supervisors Carson and Miley have endorsed Jackson. Both have been prosecutors so that's a wash. However, Jackson trumpets his social justice beliefs and given how few Black judges there are in a system bent on incarcerating young Black people,that is something. Bottom line, there are 2 good candidates to choose from.
As Central Committee members, we promote voter registration, we endorse both candidates and issues, and pass our policy ideas along to the state party in the form of resolutions. If a local candidate has been endorsed by the state party, that is us.
Given that Alameda County is one of the most progressive counties in the country, we should have an activist party that pushes the state toward more progressive solutions but that is not always so. Please consider joining, or visiting or running for one of these offices in the future.
I am a member of this body and am running again. As the local politics coordinator for the Wellstone Democratic Renewal Club and the political action chair for the Block by Block Organizing Network, BBBON, I have been very active in the struggle for affordable housing and the fight to bring an independent police commission to the City of Oakland.
Besides working for tougher renter protections (please sign the petition for the initiative proposed by the Oakland Tenants Union and ACCE, ) my next project will be working to repeal the Costa Hawkins Act which restricts any kind of renter protections on units built in the last couple of decades and into the future.
You will hear more about this next push (repeal of Costa Hawkins) in the near future. If we do not build more affordable housing while protecting those who live and work in California now, we will ultimately damage the long term economic viability of our state.
These folks are also on the slate that Rob Bonta has endorsed (also Nate Miley, Keith Carson and Abel Guillen) and while we may differ in some areas, overall we are working together to strengthen the progressive direction of our state-Robin Torello, Jim Oddie, Howard Egerman, Malcolm Amado Uno, Linda Perry, Diana Prola, Marlon McWilson (also running for County Board of Ed,) and Corina Lopez. Our entire slate has also been endorsed by the Alameda Labor Council, ALC.
My personal endorsements include-the East Bay Women's Political Caucus, Black Women Organized for Political Action, the Asian Pacific American Democratic Caucus, the John George Democratic Club, and of course, the Wellstone Democratic Renewal Club.
Yes on Measure AA-our bay needs us and wetlands are not only important to the critters who live there but as a buttress against sea level rise. It's $12 a year, folks, so while it may not be a perfect measure, it's damn good.
No.This was written after some high profile cases of corruption involving state legislators surfaced. It will do little if anything to prevent that while it damages the almost forgotten tenet that you are innocent until proven guilty. It will win but I'm not voting for it.
I'll be happy to hear your suggestion, differences, additions. Please feel free to comment!
One more thing, if someone darkens your door promoting their candidate or issue, or, heaven forfend, t calls you on your own little phone, please thank them for volunteering to promote democracy, don't yell at them for violating your privacy or interrupting the game. You might even join them next time.
4 Comments on "Pamela's Primary Voter Guide for the 18th AD"
Thanks! I was just thinking that I need to get more information about the rest of the ballot.
Thank you, Pamela, for the endless meetings you attend, speaking up, knocking doors, doing the research, encouraging all our friends to put some juice behind the word democracy. Keep on pushing.
Thank you for publishing this! My friends and I have been searching for more information on local races and this helps! I'll be voting for you!


text:
Fear The Walking Dead Showrunners On That Shocking Walker Twist
Ryan Green/AMC
By Sabienna Bowman/Nov. 23, 2020 4:11 pm EST
Over the years, The Walking Dead universe has presented viewers with an abundance of creepy walkers. From the walker in the well way back in season 2 of The Walking Dead to season 7's spike zombie, the franchise loves to get creative with the undead, but the production team may have outdone themselves with Fear the Walking Dead's season 6 midseason finale.
For the first time ever, the franchise introduced taxidermy walkers, as created by a modern day Dr. Frankenstein known as Ed. Alicia and Charlie first encounter the augmented zombies as they approach an isolated cabin, and they are a gnarly looking bunch. They later find out that the cabin's inhabitant, Ed, made the walkers more fearsome by adding horns and other frightening enhancements in hopes of protecting his family. Instead of deterring strangers, one of his creations ended up killing his loved ones.
In an interview with TV Insider, Fear the Walking Dead showrunners Andrew Chambliss and Ian Goldberg opened up about the inspiration behind the latest walker twist, sharing that the idea of taxidermy zombies has been knocking around the writers room for quite some time. "It was an idea we'd had for a while, and we were just kind of looking for the right character and episode to marry it to," said Chambliss.
Ed ended up being that character, and he has a creepy origin story that's all his own.
The Fear the Walking Dead showrunners gave the creator of the taxidermy walkers a meaningful name
During a post-show interview with Entertainment Weekly, Chambliss and Goldberg revealed that both the character of Ed and the taxidermy idea had been around for some time. Once they decided to combine the two in one episode, the show's macabre twist was born, and as an added bonus for horror fans, they named Ed after a real life serial killer. "We named the character Ed after Ed Gein, the real-life serial killer that inspired Texas Chainsaw Massacre and Silence of the Lambs, so that's just a little Easter egg," Goldberg said.
As for those nods to Mary Shelley's Frankenstein, they were entirely intentional. Chambliss explained to EW that they wanted something unique to differentiate Ed's story from all of the other stories about people who hole up alone during the apocalypse. That's how they came about not only the taxidermy idea, but also the inspiration for Ed's creations to be responsible for the death of his loved ones.
"I think what we really liked about it thematically was for Ed's story, it was all about him inadvertently creating these monsters to protect the people he loved, which ultimately destroyed the people he loved," Chambliss said. "And thematically, that really fit in with what Virginia was doing to her sister and to all the people that she's got living in her communities."
And thus a new breed of walkers was born — one that will surely be haunting fans' dreams for years to come.


Pastor Dan Kohn returns with his passion for writing. He has written daily devotions for many years and currently writes a daily devotion for Wisconsin Christian News, entitled "Story Line – God's & Ours." Pastor Dan will speak on the logistics of devotion writing, describing how a writer goes from information to inspiration through one-minute devotions.
Dan Kohn is a husband, father, pastor, and writer – in that order. His writing "career" includes five published books and devotional writing for five years. As a pastor, he develops weekly Bible studies er his goal is inspiration and application of the Word.
Awarding-winning children's author Sandia Kosmo will present her book Wakanda Whooper: The Curious Cinnamon Crane. This delightful book has received the Gold Medal Award for environmental issues and a First Place for historical fiction from Purple Dragonfly Book Awards. It takes place in Wisconsin's Necedah National Wildlife Refuge and is the story of a whooping crane. Sandia will share her writing process with the group.
Sandia Kosmo is a writer whose passion for prose and poetry has taken her many directions. She is an environmentalist who has served as a journalist, high school and college teacher, PR specialist, college administrator, and ordained pastor. After several years in New Mexico, Kosmo retired and returned to Wisconsin, but every winter, just like the whooping cranes, she migrates south.
Bill Callaghan is a father, husband, teacher, inventor, and now author. Above all, he is just a kid at heart. Bill was born in Milwaukee and grew up with ten brothers and sisters. He attended UWEC and then joined the family's mixing business in 1980 – Continental Products, Corp. For the last thirty years, he worked with his parents and four brothers, as they manufactured blending and coating equipment. Bill and his wife reside in Eau Claire with their four children.
It's that time again – time to celebrate the birth of our Lord and encourage one another to use the gifts He has given us. Join us for an informal, fun-filled evening. Laughter and chatter will come with the annual exchange of gently used books, and snacks and inspiration will accompany readings from our members. Bring a friend; bring a story; bring a snack and enjoy time with fellow writers.
A new year brings new goals. How do we set goals? What is reasonable? What is pie in the sky inconceivable? Without a vision, our writing falters, so this is the month to brainstorm, share our ideas, and make a plan. Want accountability? We can do that. Want to work in secret? That is also possible. Just put your thinking caps on, bring pen or pencil, and start your writing year with clear direction.
What do all writers recommend? Critique! How do you improve as a writer? Critique! How do I know if my message is clear? Critique! This is the month to learn about this valuable tool. First, how to do it. Then, do it. Bring your work, share in the process, and tweak your words to raise your writing to a higher level.
Life is a gift and each day we unwrap it anew. At 75, Dennis Miller is acutely mindful of God's hand over his life. From every creative thought to each breath, His calling gives him purpose. While many seek to know how to write a book, few ask why: Why write about your local history? Why write about your grief? Why write about your life? By sharing his books, Dennis Miller will encourage all to capture those special moments and snatches of time that not only live on in remembrances but transform our lives too.
Dennis R. Miller is the author of 12 books and creator/producer of 7 films. As a bereaved father, he founded Tears of Love Ministry – www.tearsofloveministry.org. He is co-founder and member of Chippewa Valley Local Authors and member of Chippewa Valley Evangelical Ministerial Association. Dennis also volunteers for the Good News Jail Ministry.
As author of Max Your Mind: The Owner's Guide for a Strong Brain, Sandra will present "The Take Away: The making of a good read – even in nonfiction." She will address the inspiration, research, and plan involved in the writing of nonfiction while she shares the honing of the craft, revision, and critiques. Submission, marketing, and publication are also in her expertise. Come ready to learn and prepared with questions. Find Sandra at Connections of the Heart LLC., or www.SandraStantonAuthor.com.
Sandra Stanton is an author, speaker, board certified coach, retired school counselor, mom, grandma and recent widow who translates brain research into user-friendly language with a side of humor. She's been fascinated by the brain since she was 14 – how it works, why it sometimes doesn't, and how to help it stay healthy. She enjoys speaking to groups of all ages through her business Connections of the Heart LLC. Her official handle is Sandra Sunquist Stanton NCC, LPC, BCC, which translates to Nationally Certified Counselor, Licensed Professional Counselor and Board Certified Coach.
Members will once again be the entertainment at our Annual Writers Recital. Invite your family and friends for an evening of creative readings and yummy refreshments. Members will describe and show their work with lots of encouragement for others. Guests will be able to browse through books and articles, and all will have time to share special writing experiences.


text:
Home » Island Hopping » Reopening event to take flight at American Airpower Museum
Grumman TBM Avenger (Photo: American Airpower Museum)
Reopening event to take flight at American Airpower Museum
July 17, 2020 July 19, 2020 The Long Island Times
The American Airpower Museum, which had been closed for more than three months due to the coronavirus, will hold a special reopening celebration next month.
The museum's reopening event takes off at 11 a.m. on Saturday, August 1, when World War II-era planes and other vintage aircraft take to the skies from the museum's ramp at Republic Airport in Farmingdale, according to a museum statement. The flights will feature a Grumman TBM Avenger, two North American T6 Texans, the AT28D5 Vietnam era combat fighter, the WACO Biplane and as an added attraction, L-39 cold-war era Russian jets.
Public health and safety concerns surrounding the pandemic led the museum's board of directors to close the museum on March 16 for two weeks.
"The health and safety of our staff, volunteers and the public was foremost," Jeff Clyman, AAM president, said in the statement. "That's why we acted early and sent everyone home, causing a total cessation of work on our aircraft." Six days later, the state ordered all non-essential businesses statewide to close, and the museum was forced to stay shut.
Admission for adults is $13, seniors and veterans $10 and children $8. Due to the need for social distancing, admission will be limited to first come/first served and a maximum attendance of 150 people will be allowed on the outdoor ramp area, with limited access to the museum.
All visitors will be required to wear face masks and will have their temperatures digitally taken at the entrance. As a special promotion, the first 20 people admitted will be included in a raffle for WACO Biplane flights later in the summer (limit one per family).
Young artists called on to compete in World Wildlife Day contest January 14, 2021
State launches offshore wind training school January 14, 2021
Westbury unveils renovations at its Community Center
October 1, 2020 The Long Island Times Comments Off on Westbury unveils renovations at its Community Center
New Yorkers can vote on new state license plate
August 19, 2019 The Long Island Times Comments Off on New Yorkers can vote on new state license plate
Trick-or-treat alley comes to Smith Haven Mall drive-in
October 16, 2020 The Long Island Times Comments Off on Trick-or-treat alley comes to Smith Haven Mall drive-in


text:
La Lega Democratica della Dardania (in albanese Lidhja demokratike e Dardanisë) è stata un partito politico del Kosovo. Il partito venne istituito nel gennaio 2007 dall'ex Presidente dell'Assemblea del Kosovo Nexhat Daci in seguito al suo tentativo fallito di diventare leader della Lega Democratica del Kosovo. Si tratta di un partito conservatore e liberal-conservatore e il secondo più grande partito di destra in Kosovo.

Il leader della Lega Democratica di Dardania era Nexhat Daci, i suoi vice erano Besa Gaxherri, Adem Salihaj, Xhemajl Hyseni e Berim Ramosaj, mentre il Segretario Generale era Lulëzim Zeneli.

La Lega Democratica ottenne seggi nell'Assemblea del Kosovo dopo la loro defezione dalla Lega Democratica del Kosovo.

Il partito fu inizialmente chiamato "Lega Democratica", ma in seguito fu aggiunto "della Dardania", l'antico nome per il Kosovo, per evitare confusione con la Lega Democratica del Kosovo.

Alle elezioni del 2010 ottiene 14 924 voti (2,14%), senza eleggere deputati.

Note

Collegamenti esterni


Fayetteville is clearly on the move, and with it former Fayette County Development Authority (FCDA) Vice President of Economic Development Emily Poole has been hired as the city's new economic development director.
A lifelong resident of Fayette County, Poole previously worked with FCDA and as public relations manager at Allan Vigil Ford. Those experiences, and her knowledge of the city and county, serve as the foundation for her new job as Fayetteville economic development director.
Poole becoming a part of the Fayetteville team is significant for the city, said City Manager Ray Gibson.
In Poole's mind, economic development is something that impacts not just businesses, but the entire community.
Poole in her new position will work with Assistant Director of Economic Development Brian Wismer, who will continue to work with the city's Main Street program, with the Fayetteville Downtown Development Authority and on special projects.
Her knowledge of the city, its past and what is unfolding in its future gives Poole a comprehensive perspective on the city. Aside from the job, Poole said that, as a homeowner, she is personally invested in the community.
"I am a lifelong resident of Fayette County. Having grown up here, I've seen the community change in so many ways, which provides me the lens of historical background and experience with our community and its people," she said.
"I truly love Fayetteville. I am passionate about making an already-wonderful community even better, and helping to build a better life for our residents. I've worked in this community for most of my professional career, and have had experience in working with a variety of industries both to support existing companies, as well as new recruitment projects. I'm pleased to bring my passion, background, partnerships and knowledge to the table to continue to grow Fayetteville," Poole said.
Awarded Fayette County Young Professional of the Year for 2011, Poole has a record of community service and partnership. The 2012-2013 President of Young Professionals of Fayette County, Poole has in the past served on the board of the Fayette Chamber of Commerce and Main Street Fayetteville. Currently, she serves on the Board of Directors for Bloom, AVPride, and Fayette Senior Services. Additionally, she is a 2013 graduate of the Atlanta Regional Commission's Regional Leadership Institute, and a 2014 graduate of Leadership Fayette.
Poole graduated from McIntosh High School in Peachtree City. She received a Bachelor's degree in Psychology and Human Services and an MBA in International Business from Clayton State University where she was recognized with the "MBA Service and Leadership" Award for the Class of 2016.