allow support for parity '0', '1' enabling support for 2,3 drive setups (#15171)

allows for further granular setups

- 2 drives (1 parity, 1 data)
- 3 drives (1 parity, 2 data)

Bonus: allows '0' parity as well.

cmd/endpoint-ellipses.go

@@ -42,7 +42,7 @@ type endpointSet struct {
 
 // Supported set sizes this is used to find the optimal
 // single set size.
-var setSizes = []uint64{4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}
+var setSizes = []uint64{2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}
 
 // getDivisibleSize - returns a greatest common divisor of
 // all the ellipses sizes.

cmd/endpoint-ellipses_test.go

@@ -201,18 +201,6 @@ func TestGetSetIndexes(t *testing.T) {
 		success    bool
 	}{
 		// Invalid inputs.
-		{
-			[]string{"data{1...3}"},
-			[]uint64{3},
-			nil,
-			false,
-		},
-		{
-			[]string{"data/controller1/export{1...2}, data/controller2/export{1...4}, data/controller3/export{1...8}"},
-			[]uint64{2, 4, 8},
-			nil,
-			false,
-		},
 		{
 			[]string{"data{1...17}/export{1...52}"},
 			[]uint64{14144},
@@ -220,6 +208,18 @@ func TestGetSetIndexes(t *testing.T) {
 			false,
 		},
 		// Valid inputs.
+		{
+			[]string{"data{1...3}"},
+			[]uint64{3},
+			[][]uint64{{3}},
+			true,
+		},
+		{
+			[]string{"data/controller1/export{1...2}, data/controller2/export{1...4}, data/controller3/export{1...8}"},
+			[]uint64{2, 4, 8},
+			[][]uint64{{2}, {2, 2}, {2, 2, 2, 2}},
+			true,
+		},
 		{
 			[]string{"data{1...27}"},
 			[]uint64{27},

cmd/erasure-metadata.go

@@ -417,7 +417,7 @@ func objectQuorumFromMeta(ctx context.Context, partsMetaData []FileInfo, errs []
 	}
 
 	parityBlocks := globalStorageClass.GetParityForSC(latestFileInfo.Metadata[xhttp.AmzStorageClass])
-	if parityBlocks <= 0 {
+	if parityBlocks < 0 {
 		parityBlocks = defaultParityCount
 	}
 

cmd/erasure-multipart.go

@@ -290,7 +290,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 
 	onlineDisks := er.getDisks()
 	parityDrives := globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
-	if parityDrives <= 0 {
+	if parityDrives < 0 {
 		parityDrives = er.defaultParityCount
 	}
 

cmd/erasure-object.go

@@ -736,7 +736,7 @@ func (er erasureObjects) putMetacacheObject(ctx context.Context, key string, r *
 	storageDisks := er.getDisks()
 	// Get parity and data drive count based on storage class metadata
 	parityDrives := globalStorageClass.GetParityForSC(opts.UserDefined[xhttp.AmzStorageClass])
-	if parityDrives <= 0 {
+	if parityDrives < 0 {
 		parityDrives = er.defaultParityCount
 	}
 	dataDrives := len(storageDisks) - parityDrives
@@ -885,7 +885,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 	if !opts.MaxParity {
 		// Get parity and data drive count based on storage class metadata
 		parityDrives = globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
-		if parityDrives <= 0 {
+		if parityDrives < 0 {
 			parityDrives = er.defaultParityCount
 		}
 

cmd/erasure-object_test.go

@@ -893,6 +893,14 @@ func testObjectQuorumFromMeta(obj ObjectLayer, instanceType string, dirs []strin
 
 	// Object for test case 1 - No StorageClass defined, no MetaData in PutObject
 	object1 := "object1"
+	globalStorageClass = storageclass.Config{
+		RRS: storageclass.StorageClass{
+			Parity: 2,
+		},
+		Standard: storageclass.StorageClass{
+			Parity: 4,
+		},
+	}
 	_, err = obj.PutObject(ctx, bucket, object1, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), opts)
 	if err != nil {
 		t.Fatalf("Failed to putObject %v", err)
@@ -964,17 +972,16 @@ func testObjectQuorumFromMeta(obj ObjectLayer, instanceType string, dirs []strin
 	}
 
 	parts5, errs5 := readAllFileInfo(ctx, erasureDisks, bucket, object5, "", false)
-	parts5SC := storageclass.Config{
-		RRS: storageclass.StorageClass{
-			Parity: 2,
-		},
-	}
+	parts5SC := globalStorageClass
 
 	// Object for test case 6 - RRS StorageClass defined as Parity 2, MetaData in PutObject requesting Standard Storage Class
 	object6 := "object6"
 	metadata6 := make(map[string]string)
 	metadata6["x-amz-storage-class"] = storageclass.STANDARD
 	globalStorageClass = storageclass.Config{
+		Standard: storageclass.StorageClass{
+			Parity: 4,
+		},
 		RRS: storageclass.StorageClass{
 			Parity: 2,
 		},
@@ -1035,7 +1042,7 @@ func testObjectQuorumFromMeta(obj ObjectLayer, instanceType string, dirs []strin
 		tt := tt
 		t.(*testing.T).Run("", func(t *testing.T) {
 			globalStorageClass = tt.storageClassCfg
-			actualReadQuorum, actualWriteQuorum, err := objectQuorumFromMeta(ctx, tt.parts, tt.errs, getDefaultParityBlocks(len(erasureDisks)))
+			actualReadQuorum, actualWriteQuorum, err := objectQuorumFromMeta(ctx, tt.parts, tt.errs, storageclass.DefaultParityBlocks(len(erasureDisks)))
 			if tt.expectedError != nil && err == nil {
 				t.Errorf("Expected %s, got %s", tt.expectedError, err)
 			}

cmd/erasure-server-pool.go

@@ -525,7 +525,7 @@ func (z *erasureServerPools) BackendInfo() (b madmin.BackendInfo) {
 	b.Type = madmin.Erasure
 
 	scParity := globalStorageClass.GetParityForSC(storageclass.STANDARD)
-	if scParity <= 0 {
+	if scParity < 0 {
 		scParity = z.serverPools[0].defaultParityCount
 	}
 	rrSCParity := globalStorageClass.GetParityForSC(storageclass.RRS)

cmd/format-erasure.go

@@ -643,7 +643,7 @@ func saveFormatErasureAll(ctx context.Context, storageDisks []StorageAPI, format
 		}, index)
 	}
 
-	writeQuorum := getWriteQuorum(len(storageDisks))
+	writeQuorum := (len(storageDisks) + 1/2)
 	// Wait for the routines to finish.
 	return reduceWriteQuorumErrs(ctx, g.Wait(), nil, writeQuorum)
 }
@@ -805,26 +805,13 @@ func initFormatErasure(ctx context.Context, storageDisks []StorageAPI, setCount,
 	return getFormatErasureInQuorum(formats)
 }
 
-func getDefaultParityBlocks(drive int) int {
-	switch drive {
-	case 3, 2:
-		return 1
-	case 4, 5:
-		return 2
-	case 6, 7:
-		return 3
-	default:
-		return 4
-	}
-}
-
 // ecDrivesNoConfig returns the erasure coded drives in a set if no config has been set.
 // It will attempt to read it from env variable and fall back to drives/2.
 func ecDrivesNoConfig(setDriveCount int) int {
 	sc, _ := storageclass.LookupConfig(config.KVS{}, setDriveCount)
 	ecDrives := sc.GetParityForSC(storageclass.STANDARD)
-	if ecDrives <= 0 {
-		ecDrives = getDefaultParityBlocks(setDriveCount)
+	if ecDrives < 0 {
+		ecDrives = storageclass.DefaultParityBlocks(setDriveCount)
 	}
 	return ecDrives
 }

cmd/utils.go

@@ -121,15 +121,6 @@ func path2BucketObject(s string) (bucket, prefix string) {
 	return path2BucketObjectWithBasePath("", s)
 }
 
-func getWriteQuorum(drive int) int {
-	parity := getDefaultParityBlocks(drive)
-	quorum := drive - parity
-	if quorum == parity {
-		quorum++
-	}
-	return quorum
-}
-
 // CloneMSS is an exposed function of cloneMSS for gateway usage.
 var CloneMSS = cloneMSS